如何将AWS MSK Kafka数组数据转为Snowflake中的独立行？

一、Kafka端预处理（推荐，减少Snowflake端计算开销） #

1. 使用Kafka Connect单消息转换（SMT）拆分数组 #

Snowflake Kafka Connector支持通过SMT直接拆分数组，无需额外代码。在连接器配置中添加以下参数，将顶层数组的每个元素拆分为独立的Kafka消息：

# 启用数组拆分转换
transforms=expandArray
transforms.expandArray.type=org.apache.kafka.connect.transforms.ExpandArray$Value

配置完成后，连接器会自动将每个数组对象转为单独的消息，写入Snowflake时每条消息对应一行记录。

2. 用Kafka Streams提前拆分数组 #

如果需要更灵活的处理逻辑，可以编写轻量Kafka Streams应用，消费原始Topic并将数组元素拆分后发送到新Topic，再让Snowflake连接器消费新Topic。示例Java代码：

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.ArrayList;
import java.util.List;

public class ArraySplitStream {
    public static void main(String[] args) {
        StreamsConfig config = new StreamsConfig(StreamsUtils.loadConfig());
        StreamsBuilder builder = new StreamsBuilder();
        ObjectMapper mapper = new ObjectMapper();

        KStream<String, String> input = builder.stream("original-topic");
        KStream<String, String> output = input.flatMapValues(rawJson -> {
            List<String> result = new ArrayList<>();
            try {
                JsonNode arrayNode = mapper.readTree(rawJson);
                if (arrayNode.isArray()) {
                    for (JsonNode node : arrayNode) {
                        result.add(node.toString());
                    }
                }
            } catch (Exception e) {
                // 处理解析异常
                e.printStackTrace();
            }
            return result;
        });

        output.to("flattened-topic");
        KafkaStreams streams = new KafkaStreams(builder.build(), config);
        streams.start();
    }
}

二、Snowflake端后处理方案 #

若无法修改Kafka端配置，可先将数组数据加载到临时表，再用FLATTEN函数拆分：

1. 创建临时表接收原始数组数据：

CREATE OR REPLACE TEMPORARY TABLE raw_kafka_data (
    payload ARRAY
);

2. 配置Snowflake连接器将数据写入上述临时表（确保字段映射正确）。
3. 拆分数据并插入目标表：

CREATE OR REPLACE TABLE target_booster_data (
    ID NUMBER,
    "Booster Station" VARCHAR,
    "TimeStamp" TIMESTAMP_NTZ,
    "Total Throughput" FLOAT
);

INSERT INTO target_booster_data
SELECT
    value:Id::NUMBER,
    value:"Booster Station"::VARCHAR,
    value:"TimeStamp"::TIMESTAMP_NTZ,
    value:"Total Throughput"::FLOAT
FROM raw_kafka_data,
LATERAL FLATTEN(input => payload);

如果需要自动同步，可创建Snowflake任务定期执行上述插入逻辑。

内容的提问来源于stack exchange，提问作者Mohammed Ali