如何在KSQL持久化查询流中提取MAP结构的指定键值？

当然可以实现，Confluent旗下的KSQL和Kafka Streams API都支持这种复杂类型的字段提取操作，以下是两种常用方案：

方案一：使用KSQL（适合无代码/低代码场景） #

先基于你的数据结构创建输入流，指定Field1为包含STRUCT值的MAP类型，再通过键访问语法提取目标属性：

1. 创建输入流 #

CREATE STREAM input_stream (
    Field1 MAP<VARCHAR, STRUCT<val1 INT, val2 INT, val3 INT>>
) WITH (
    KAFKA_TOPIC='your_source_topic',
    VALUE_FORMAT='JSON',
    PARTITIONS=6 -- 根据你的实际topic配置调整
);

2. 提取独立属性到新流 #

如果只需要提取整个嵌套结构作为独立字段：

CREATE STREAM output_stream AS
SELECT
    Field1['varchar_value_i_want'] AS extracted_target1,
    Field1['varchar_value_i_also_want'] AS extracted_target2,
    Field1 -- 可选：保留原字段
FROM input_stream;

如果需要进一步拆分嵌套STRUCT里的子字段：

CREATE STREAM output_stream AS
SELECT
    -- 拆分第一个目标属性的子字段
    Field1['varchar_value_i_want']->val1 AS target1_val1,
    Field1['varchar_value_i_want']->val2 AS target1_val2,
    Field1['varchar_value_i_want']->val3 AS target1_val3,
    -- 拆分第二个目标属性的子字段
    Field1['varchar_value_i_also_want']->val1 AS target2_val1,
    Field1['varchar_value_i_also_want']->val2 AS target2_val2,
    Field1['varchar_value_i_also_want']->val3 AS target2_val3
FROM input_stream;

KSQL 5.3及以上版本（对应Confluent Platform 5.3+）就支持这种MAP键访问和STRUCT字段提取操作，当前主流的Confluent稳定版（7.x、8.x）完全兼容。

方案二：使用Kafka Streams API（适合自定义代码场景） #

如果需要更灵活的自定义处理，可以用Java/Python等语言的Streams API实现，以下是Java示例：

1. 定义数据模型 #

假设用POJO作为数据载体（也可以用Avro/Protobuf配合Schema Registry）：

// 输入数据模型
public class InputData {
    private Map<String, NestedData> Field1;
    // getter、setter省略
}

// 嵌套结构模型
public class NestedData {
    private int val1;
    private int val2;
    private int val3;
    // getter、setter省略
}

// 输出数据模型（包含独立属性）
public class OutputData {
    private NestedData target1;
    private NestedData target2;
    // 构造方法、getter、setter省略
}

2. 编写流处理逻辑 #

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class MapExtractionExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, InputData> inputStream = builder.stream("your_source_topic");

        // 提取目标属性并构造输出
        KStream<String, OutputData> outputStream = inputStream.mapValues(input -> {
            Map<String, NestedData> field1 = input.getField1();
            NestedData target1 = field1.get("varchar_value_i_want");
            NestedData target2 = field1.get("varchar_value_i_also_want");
            return new OutputData(target1, target2);
        });

        // 将结果输出到目标topic
        outputStream.to("your_output_topic");
    }
}

这种方式不受版本限制，只要是支持Kafka Streams的Confluent版本都能运行，配合Schema Registry可以更高效地处理序列化/反序列化。

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