如何用不同Key关联Kafka流与其他Topic实现数据补全？

你的核心问题是需要基于keyval字段关联S流和T1流的所有匹配记录，同时保留T1的全量历史记录（而非最新一条），还要保证消费所有分区。下面是具体的实现方案：

1. 核心思路 #

• 放弃使用KTable/GlobalKTable：这类组件会按Key去重，只保留最新记录，不符合你需要全量匹配的需求。
• 将T1转为KStream，提取keyval作为新Key，聚合到持久化状态存储中，存储每个keyval对应的所有T1记录列表。
• 对S流提取keyval后，通过状态存储查询所有匹配的T1记录，完成关联。

2. Spring Cloud Stream具体实现 #

配置文件（application.yml） #

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: your-kafka-broker-list
            application-id: kafka-streams-join-app
      bindings:
        s-input:
          destination: S
          group: your-consumer-group
        t1-input:
          destination: T1
          group: your-consumer-group
        output:
          destination: joined-output-topic

流处理代码 #

首先定义对应CDC同步的实体类（以Debezium风格为例）：

// S流的记录实体
@Data
public class SRecord {
    private String originalKey; // 保存原始的keyval|someval
    private String val1;
    private String val2;
}

// T1流的CDC记录实体
@Data
public class TRecord {
    private String id; // 数据库主键，用于识别更新/删除
    private String op; // CDC操作类型：C(创建)/U(更新)/D(删除)
    private String val1;
    private String val2;
    private String val3;
}

// 关联后的结果实体
@Data
public class CombinedRecord {
    private SRecord sRecord;
    private List<TRecord> matchedTRecords;
}

然后是流处理器配置：

@Configuration
public class StreamJoinConfig {
    private static final String T1_RECORDS_STORE = "t1-keyval-store";

    // 处理S流，关联T1的全量记录
    @Bean
    public Function<KStream<String, SRecord>, KStream<String, CombinedRecord>> processSStream() {
        return sStream -> {
            // 从S的Key中提取keyval部分
            KStream<String, SRecord> sWithKeyval = sStream
                    .map((key, sRecord) -> {
                        sRecord.setOriginalKey(key); // 保存原始Key
                        String keyval = key.split("\\|")[0];
                        return KeyValue.pair(keyval, sRecord);
                    });

            // 使用Transform查询状态存储，关联所有匹配的T1记录
            return sWithKeyval.transform(() -> new T1RecordJoiner(), T1_RECORDS_STORE)
                    // 还原回原始的S流Key作为结果Key（可选，根据需求调整）
                    .map((keyval, combined) -> KeyValue.pair(combined.getSRecord().getOriginalKey(), combined));
        };
    }

    // 处理T1流，聚合到状态存储
    @Bean
    public Consumer<KStream<String, TRecord>> processT1Stream(StreamsBuilder streamsBuilder) {
        // 创建持久化的状态存储，存储keyval到TRecord列表的映射
        StoreBuilder<KeyValueStore<String, List<TRecord>>> storeBuilder = Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore(T1_RECORDS_STORE),
                Serdes.String(),
                new ListJsonSerde<>(TRecord.class)
        );
        streamsBuilder.addStateStore(storeBuilder);

        return t1Stream -> {
            // 从T1的Key中提取keyval部分（原Key格式是tabval|keyval）
            KStream<String, TRecord> t1WithKeyval = t1Stream
                    .map((key, tRecord) -> {
                        String keyval = key.split("\\|")[1];
                        return KeyValue.pair(keyval, tRecord);
                    });

            // 聚合每个keyval的所有TRecord到列表，处理CDC的增删改
            t1WithKeyval.groupByKey()
                    .aggregate(
                            ArrayList::new, // 初始化空列表
                            (keyval, newTRecord, existingList) -> {
                                switch (newTRecord.getOp()) {
                                    case "D":
                                        // 删除事件：移除对应主键的记录
                                        existingList.removeIf(record -> record.getId().equals(newTRecord.getId()));
                                        break;
                                    case "C":
                                    case "U":
                                        // 新增/更新：先移除旧的同主键记录，再添加新的
                                        existingList.removeIf(record -> record.getId().equals(newTRecord.getId()));
                                        existingList.add(newTRecord);
                                        break;
                                }
                                return existingList;
                            },
                            Materialized.<String, List<TRecord>, KeyValueStore<Bytes, byte[]>>as(T1_RECORDS_STORE)
                                    .withKeySerde(Serdes.String())
                                    .withValueSerde(new ListJsonSerde<>(TRecord.class))
                    );
        };
    }

    // 自定义的List JSON Serde，用于序列化TRecord列表
    static class ListJsonSerde<T> extends Serdes.WrapperSerde<List<T>> {
        public ListJsonSerde(Class<T> elementType) {
            super(new JsonSerde<>(List.class, elementType), new JsonSerde<>(List.class, elementType));
        }
    }

    // 自定义Transformer，用于查询状态存储并关联记录
    static class T1RecordJoiner implements Transformer<String, SRecord, KeyValue<String, CombinedRecord>> {
        private KeyValueStore<String, List<TRecord>> t1Store;

        @Override
        public void init(ProcessorContext context) {
            // 获取状态存储实例
            t1Store = context.getStateStore(T1_RECORDS_STORE);
        }

        @Override
        public KeyValue<String, CombinedRecord> transform(String keyval, SRecord sRecord) {
            // 查询当前keyval对应的所有T1记录
            List<TRecord> tRecords = Optional.ofNullable(t1Store.get(keyval)).orElse(new ArrayList<>());
            CombinedRecord combined = new CombinedRecord();
            combined.setSRecord(sRecord);
            combined.setMatchedTRecords(tRecords);
            return KeyValue.pair(keyval, combined);
        }

        @Override
        public void close() {
            // 资源清理（如果需要）
        }
    }
}

3. 关键注意事项 #

• CDC事件处理：因为T1是从数据库同步的CDC流，必须处理增删改事件，避免状态存储中积累无效数据。代码中通过op字段判断操作类型，更新/删除时维护列表的正确性。
• 状态存储持久化：使用persistentKeyValueStore保证应用重启后状态不丢失，避免重新消费全量T1数据。
• 分区消费保障：Spring Cloud Stream的消费者组配置会确保Kafka Streams分配所有Topic分区给应用实例，只要应用实例数不超过分区数（或者配置合适的分区分配策略），就能消费所有分区的数据。
• Serde兼容性：自定义ListJsonSerde解决列表类型的序列化问题，也可以用Spring提供的JsonSerde直接包装。

4. 替代方案（适合有时间窗口的场景） #

如果你的业务允许只关联最近一段时间内的T1记录，可以使用KStream-KStream的窗口join：

// 示例：5分钟窗口内的join
KStream<String, CombinedRecord> joinedStream = sWithKeyval.join(
        t1WithKeyval,
        (sRecord, tRecord) -> {
            // 一对多场景下仍需先聚合T1到列表，因此这种方式更适合一对一关联
            CombinedRecord combined = new CombinedRecord();
            combined.setSRecord(sRecord);
            combined.setMatchedTRecords(Collections.singletonList(tRecord));
            return combined;
        },
        JoinWindows.of(Duration.ofMinutes(5)),
        Joined.with(Serdes.String(), new JsonSerde<>(SRecord.class), new JsonSerde<>(TRecord.class))
);

但这种方式无法保留全量历史记录，仅适合有时间范围限制的场景。

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