如何用KSQL处理字段为ARRAY或STRUCT的Kafka消息？

方案一：纯KSQL内置函数统一格式 #

先将目标字段以STRING类型捕获，通过条件判断将单个STRUCT包装为ARRAY，再统一解析为ARRAY<STRUCT>类型：

-- 1. 创建原始流，接收原始主题消息，将field字段转为STRING类型
CREATE STREAM raw_topic_stream (field STRING) 
WITH (KAFKA_TOPIC='your_source_topic', VALUE_FORMAT='JSON');

-- 2. 生成标准化流：统一所有格式为ARRAY<STRUCT>
CREATE STREAM standardized_field_stream AS
SELECT
  CASE
    -- 判断字段是否为数组格式
    WHEN LEFT(field, 1) = '[' THEN JSON_PARSE(field)
    -- 单个STRUCT包装为单元素数组后解析
    ELSE ARRAY[JSON_PARSE(field)]
  END AS standardized_field
FROM raw_topic_stream
EMIT CHANGES;

逻辑说明：通过LEFT()函数识别数组开头的[，将所有格式统一为ARRAY类型，后续可直接对standardized_field执行数组操作（如UNWIND、索引访问等）。

方案二：自定义UDF扩展KSQL能力 #

如果内置函数无法处理复杂嵌套场景，可编写自定义UDF（User-Defined Function）实现格式兼容：

1. 编写UDF核心逻辑（Java示例）：

import org.apache.kafka.connect.data.Struct;
import io.confluent.ksql.function.udf.Udf;
import io.confluent.ksql.function.udf.UdfDescription;
import java.util.ArrayList;
import java.util.List;
import com.fasterxml.jackson.databind.ObjectMapper;

@UdfDescription(name = "standardize_struct_array", description = "统一单个STRUCT或STRUCT数组为数组格式")
public class StructArrayStandardizer {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Udf(description = "将字段字符串转换为STRUCT数组")
    public List<Struct> standardize(String fieldJson) throws Exception {
        List<Struct> result = new ArrayList<>();
        if (fieldJson.startsWith("[")) {
            // 直接解析为STRUCT数组
            Struct[] structs = MAPPER.readValue(fieldJson, Struct[].class);
            for (Struct s : structs) {
                result.add(s);
            }
        } else {
            // 单个STRUCT包装为数组
            Struct struct = MAPPER.readValue(fieldJson, Struct.class);
            result.add(struct);
        }
        return result;
    }
}

2. 将UDF打包部署到Confluent平台，在KSQL中注册并调用：

CREATE FUNCTION standardize_struct_array AS 'com.your.package.StructArrayStandardizer';

CREATE STREAM standardized_stream AS
SELECT standardize_struct_array(field) AS standardized_field
FROM raw_topic_stream
EMIT CHANGES;

方案三：Kafka Streams预处理（脱离纯KSQL场景） #

若允许编写预处理应用，直接在Kafka Streams层完成格式统一，再输出到新主题供KSQL消费：

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;

public class FieldStandardizerApp {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> sourceStream = builder.stream("your_source_topic");
        
        KStream<String, JsonNode> standardizedStream = sourceStream.mapValues(value -> {
            try {
                JsonNode root = MAPPER.readTree(value);
                JsonNode fieldNode = root.get("field");
                if (fieldNode.isArray()) {
                    return fieldNode;
                } else {
                    // 单个STRUCT转为单元素数组
                    ArrayNode arrayNode = MAPPER.createArrayNode();
                    arrayNode.add(fieldNode);
                    return arrayNode;
                }
            } catch (Exception e) {
                // 自定义异常处理逻辑
                return null;
            }
        });
        
        standardizedStream.to("standardized_topic");
    }
}

部署该应用后，KSQL直接消费standardized_topic即可，无需再处理格式兼容问题。

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