如何读取嵌套AVRO字段以创建流？附Kafka Topic的AVRO消息示例

看起来你需要从嵌套结构的AVRO消息中提取字段来构建流处理任务，我来一步步帮你搞定这个问题。

第一步：明确你的AVRO Schema结构 #

首先得把你的AVRO Schema梳理清楚，从你给出的消息示例来看，完整的Schema大概是这样的（我补充了缺失的部分）：

{
  "type": "record",
  "name": "CDCEvent",
  "namespace": "com.yourcompany.cdc",
  "fields": [
    {"name": "table", "type": "string"},
    {"name": "op_type", "type": "string"},
    {"name": "op_ts", "type": "string"},
    {"name": "current_ts", "type": "string"},
    {"name": "pos", "type": "string"},
    {"name": "before", "type": ["null", "record"], "default": null, "fields": [{"name": "row", "type": "record", "fields": []}]},
    {"name": "after", "type": ["null", "record"], "default": null, "fields": [
      {"name": "row", "type": "record", "name": "DealRow", "fields": [
        {"name": "DEA_PID_DEAL", "type": "string"},
        {"name": "DEA_NME_DEAL", "type": "string"},
        {"name": "DEA_NME_ALIAS_NAME", "type": "string"},
        {"name": "DEA_NUM_DEAL_CNTL", "type": "string"}
      ]}
    ]}
  ]
}

注：这里的before和after字段是可空的record类型，对应你消息里的null值和嵌套的row结构。

第二步：选择流处理工具并解析嵌套字段 #

下面我分别给出两种常用方式的实现：Kafka Streams（Java）和KSQL（SQL方式），你可以根据自己的技术栈选择。

方式一：用Kafka Streams（Java）处理 #

假设你已经配置好了Kafka Streams的基础环境，包括Schema Registry的连接（因为是AVRO格式，推荐用Confluent Schema Registry来管理Schema）。

1. 首先定义对应的POJO类（可以用Avro工具自动生成，比如avro-maven-plugin），生成后你会得到CDCEvent、DealRow等类。

2. 在流处理代码中，提取嵌套字段：

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import com.yourcompany.cdc.CDCEvent;
import com.yourcompany.cdc.DealRow;

public class DealStreamProcessor {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        
        // 从Kafka Topic读取AVRO消息，反序列化为CDCEvent对象
        KStream<String, CDCEvent> sourceStream = builder.stream("your-topic-name");
        
        // 过滤出op_type为Insert的消息，并且after不为null的记录
        KStream<String, DealRow> dealStream = sourceStream
            .filter((key, event) -> "Insert".equals(event.getOpType()) && event.getAfter() != null)
            .mapValues(event -> event.getAfter().getRow()); // 提取after里的row字段
        
        // 现在你可以对dealStream做后续处理，比如过滤、聚合、输出到其他Topic等
        dealStream.to("processed-deal-topic");
        
        // 启动Kafka Streams应用...
    }
}

关键点：通过event.getAfter().getRow()直接访问嵌套字段，这得益于Avro生成的POJO提供了类型安全的getter方法。

方式二：用KSQL（无需代码，SQL方式） #

如果不想写代码，KSQL是更快捷的方式，前提是你的Kafka集群已经部署了Confluent Platform或者KSQL Server。

1. 首先注册AVRO格式的Topic：

CREATE STREAM cdc_deal_stream (
    table STRING,
    op_type STRING,
    op_ts STRING,
    current_ts STRING,
    pos STRING,
    before STRUCT<row:STRUCT<>>,
    after STRUCT<row:STRUCT<
        DEA_PID_DEAL STRING,
        DEA_NME_DEAL STRING,
        DEA_NME_ALIAS_NAME STRING,
        DEA_NUM_DEAL_CNTL STRING
    >>
) WITH (
    KAFKA_TOPIC='your-topic-name',
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='STRING'
);

2. 然后创建一个处理后的流，提取嵌套字段：

CREATE STREAM processed_deal_stream AS
SELECT
    after->row->DEA_PID_DEAL AS deal_id,
    after->row->DEA_NME_DEAL AS deal_name,
    after->row->DEA_NME_ALIAS_NAME AS deal_alias,
    op_ts AS operation_time
FROM cdc_deal_stream
WHERE op_type = 'Insert' AND after IS NOT NULL;

这里用->操作符来访问嵌套的STRUCT字段，非常直观。

注意事项 #

• 确保你的Schema Registry配置正确，流处理工具能自动获取或匹配AVRO Schema，避免序列化/反序列化错误。
• 如果before或after字段可能为null，一定要在代码或SQL中做非空判断，避免空指针异常。
• 如果你的AVRO Schema有变动，要注意Schema兼容性（比如用向前兼容的方式修改Schema），避免流处理任务崩溃。

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