如何使用Kafka Streams将Topic1指定键值字段推送至Topic2

核心实现逻辑 #

要实现类似ReplaceField$Whitelist的字段白名单筛选效果，不需要引入额外SMT，直接在KStream算子中构造匹配子集Schema的GenericRecord即可，核心步骤如下：

• 提前定义仅包含目标保留字段的Avro Schema，Schema中字段的名称、类型必须和原Topic对应字段完全一致
• 在流处理的map/mapValues算子中，基于该子集Schema新建GenericRecord实例
• 从原消息的键/值中逐个提取需要保留的字段，写入新的GenericRecord实例
• 配置匹配输出Schema的Avro序列化器，将处理后的流写入目标Topic

代码示例 #

1. 定义输出用的子集Schema #

比如原消息值包含id、name、address、phoneno四个字段，仅保留前三个的Schema定义如下：

{
  "type": "record",
  "name": "SimplifiedBizRecord",
  "namespace": "com.example.biz",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"},
    {"name": "address", "type": ["null", "string"], "default": null}
  ]
}

可以直接在代码中初始化该Schema，也可以提前将Schema注册到Confluent Schema Registry后在客户端拉取使用。

2. 流处理核心逻辑（Java实现） #

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;

// 初始化子集Schema
private static final Schema TARGET_VALUE_SCHEMA = new Schema.Parser().parse("/* 替换为上述Avro Schema的JSON字符串 */");

public void buildStream(StreamsBuilder builder) {
    // 接入源Topic，配置对应Avro Serde
    KStream<GenericRecord, GenericRecord> sourceStream = builder.stream(
        "Topic1",
        Consumed.with(/* 源Topic键的Avro Serde */, /* 源Topic值的Avro Serde */)
    );

    sourceStream
        // 如果需要同时修改消息键、从键中提取字段，替换为map((key, value) -> KeyValue.pair(newKey, newValue))即可
        .mapValues(sourceValue -> {
            // 新建绑定目标Schema的GenericRecord，禁止直接修改传入的sourceValue
            GenericRecord targetValue = new GenericData.Record(TARGET_VALUE_SCHEMA);
            // 逐个写入需要保留的字段
            targetValue.put("id", sourceValue.get("id"));
            targetValue.put("name", sourceValue.get("name"));
            targetValue.put("address", sourceValue.get("address"));
            return targetValue;
        })
        // 写入目标Topic，值Serde必须匹配TARGET_VALUE_SCHEMA
        .to(
            "Topic2",
            Produced.with(/* 目标Topic键的Serde */, /* 匹配子集Schema的值Avro Serde */)
        );
}

注意事项 #

• 传入map/mapValues的原消息键、值对象是只读的，禁止直接修改原对象的字段，必须新建GenericRecord实例承载筛选后的字段，否则会抛出运行时异常
• 新构造GenericRecord绑定的Schema，必须和输出端配置的Avro Serde使用的Schema完全匹配，字段顺序、类型、名称不能有差异，否则会触发序列化报错
• 如果保留的字段包含嵌套结构、数组、Map等复杂类型，直接通过get()方法取出后放入新Record即可，不需要额外做类型转换，只要两边Schema的字段类型一致就能正常序列化
• 如果需要从原消息键中提取字段合并到值中，只需要在map算子中同时读取key、value的字段，按上述逻辑构造新值即可

内容的提问来源于stack exchange，提问作者Abhijeet Singh Rana