基于Apache Kafka合并单主题单事务多JSON为嵌套JSON的方案咨询

核心工具选择：Kafka Streams #

Kafka Streams是Kafka生态原生的流处理库，完美适配你这种同一事务内多消息聚合的场景，无需引入外部依赖，直接基于Kafka集群运行。

实现思路 #

1. 按事务ID分组：确保每条消息包含唯一transaction_id（无此字段则需通过事务标识字段生成），通过groupByKey()将同一事务的所有消息聚合到同一处理流。
2. 状态存储累积消息：用Kafka Streams的**键值状态存储（KeyValueStore）**临时保存事务内各表消息，直到收到status: "END"的消息。
3. 合并嵌套JSON：收到END消息时，从状态存储取出该事务的所有表数据，组装成嵌套JSON结构后发送到目标主题。
4. 清理状态：发送完成后删除状态存储中的临时数据，避免内存占用。

关键代码示例 #

假设输入消息结构如下：

{
  "transaction_id": "txn_12345",
  "table": "Customer_details",
  "status": "PROCESSING",
  "data": { "customer_id": "C001", "name": "John Doe" }
}

Kafka Streams核心处理逻辑：

// 初始化配置
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "transaction-json-aggregator");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

// 构建流拓扑
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> orderStream = builder.stream("Orders");

// 按transaction_id分组
KStream<String, String> keyedStream = orderStream.map((key, value) -> {
    JsonNode json = new ObjectMapper().readTree(value);
    String txnId = json.get("transaction_id").asText();
    return KeyValue.pair(txnId, value);
});

// 创建持久化状态存储
StoreBuilder<KeyValueStore<String, String>> storeBuilder =
        Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore("transaction-store"),
                Serdes.String(),
                Serdes.String());
builder.addStateStore(storeBuilder);

// 处理消息并聚合
keyedStream.process(() -> new Processor<String, String>() {
    private KeyValueStore<String, String> store;
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public void init(ProcessorContext context) {
        store = context.getStateStore("transaction-store");
    }

    @Override
    public void process(String txnId, String value) {
        JsonNode msgNode = mapper.readTree(value);
        String status = msgNode.get("status").asText();
        String table = msgNode.get("table").asText();
        JsonNode data = msgNode.get("data");

        // 初始化或更新嵌套结构
        String existingData = store.get(txnId);
        JsonNode mergedJson = existingData == null ? mapper.createObjectNode() : mapper.readTree(existingData);
        ((ObjectNode) mergedJson).set(table, data);

        if ("BEGIN".equals(status)) {
            store.put(txnId, mergedJson.toString());
        } else if ("END".equals(status)) {
            // 发送合并结果到目标主题
            context.forward(txnId, mergedJson.toString());
            // 清理临时数据
            store.delete(txnId);
        } else {
            store.put(txnId, mergedJson.toString());
        }
    }

    @Override
    public void close() {}
}, "transaction-store");

// 输出到目标主题
keyedStream.to("Merged_Orders");

// 启动流应用
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

注意事项 #

• 消息顺序：通过将transaction_id设为分区键，保证同一事务的消息按顺序到达，避免合并逻辑出错。
• 状态持久化：使用持久化存储可避免应用重启时丢失未完成的事务数据。
• 超时清理：添加定时器清理状态存储中超时的未完成事务（如超过30分钟未收到END的事务），防止内存泄漏。

无代码替代方案：KSQL #

若不想编写代码，可使用KSQL（Kafka SQL流处理引擎）实现聚合，示例语句：

CREATE STREAM merged_orders AS
SELECT transaction_id,
       COLLECT_LIST(STRUCT(table_name := table, data := data)) AS all_tables
FROM Orders
GROUP BY transaction_id
EMIT FINAL WHEN status = 'END';

将查询结果导出到目标主题即可。

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