Kafka Streams：合并同Key双流并按时间关联温度与事件类型

要实现将温度数据关联到其前一个事件类型的需求，核心思路是用KTable持久化最新的事件状态，再将温度流与这个状态表做关联，具体步骤如下：

1. 将事件流转换为KTable保存最新状态 #

因为所有记录的Key都是station，KTable会自动保留该Key对应的最新事件值（也就是最近一次的red/green），后续温度流可以直接获取这个最新状态：

// 将Stream2转换为KTable，保存最新的事件类型
KTable<String, String> latestEventTable = Stream2.toTable();

2. 温度流与事件表做左连接 #

使用leftJoin操作，让每个温度记录关联当前KTable中最新的事件值——也就是该温度发生之前最近的事件类型。这里需要注意：

• 左连接会保留所有温度记录，即使之前没有任何事件（比如前两条温度-35º、-40º，此时事件表为空，关联结果为null，可根据需求设置默认值）
• Kafka Streams会按事件时间自动处理顺序，只要生产者发送记录时携带了正确的时间戳（如你给出的时序），就能保证关联的是前一个事件

// 温度流左连接事件表，关联最近的事件类型
KStream<String, String> joinedStream = Stream1.leftJoin(
    latestEventTable,
    (temperature, latestEvent) -> {
        // 处理关联结果：无前置事件时用默认值"none"
        String eventType = (latestEvent == null) ? "none" : latestEvent;
        return String.format("温度: %s，关联事件: %s", temperature, eventType);
    }
);

// 将结果输出到目标topic
joinedStream.to(outputTopic);

3. 确保时间顺序的关键配置 #

如果生产者发送记录时未携带正确的事件时间，需显式设置TimestampExtractor来指定时间戳字段，保证Kafka Streams按事件时间处理：

Properties props = new Properties();
// 自定义时间戳提取器，需实现TimestampExtractor接口，根据实际数据格式解析时间
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, CustomTimestampExtractor.class);

StreamsBuilder builder = new StreamsBuilder();
// 后续创建流的代码...

输出结果验证 #

根据你给出的时序，最终输出会是：

• 温度: -35º，关联事件: none
• 温度: -40º，关联事件: none
• 温度: 20º，关联事件: red
• 温度: 10º，关联事件: green
• 温度: 40º，关联事件: red

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