如何关联键近似相同的两个Java KStream（Kafka Topic）？

你的问题核心是原Topic的键包含不需要匹配的字段（row），导致直接用innerJoin无法按预期关联。解决思路是先重新构造用于关联的复合键，完成join后再恢复原Topic A的键，具体步骤如下：

一、核心思路 #

要匹配的条件是：serial、date、machine（来自原键） + no（来自value），所以需要把这四个字段组合成新的关联键，让两个Topic的流基于这个新键做join，之后再把结果的键换回Topic A的原键。

二、Kafka Streams代码实现（Java） #

1. 定义复合关联键类 #

这个类要包含所有用于匹配的字段，并且必须实现equals()和hashCode()方法，否则Kafka无法正确分组匹配：

import java.util.Objects;

public class JoinKey {
    private Integer serial;
    private String date;
    private String machine;
    private Integer no;

    // 全参构造函数
    public JoinKey(Integer serial, String date, String machine, Integer no) {
        this.serial = serial;
        this.date = date;
        this.machine = machine;
        this.no = no;
    }

    // getter和setter方法
    public Integer getSerial() { return serial; }
    public void setSerial(Integer serial) { this.serial = serial; }
    public String getDate() { return date; }
    public void setDate(String date) { this.date = date; }
    public String getMachine() { return machine; }
    public void setMachine(String machine) { this.machine = machine; }
    public Integer getNo() { return no; }
    public void setNo(Integer no) { this.no = no; }

    // 必须实现equals和hashCode
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        JoinKey joinKey = (JoinKey) o;
        return Objects.equals(serial, joinKey.serial) &&
               Objects.equals(date, joinKey.date) &&
               Objects.equals(machine, joinKey.machine) &&
               Objects.equals(no, joinKey.no);
    }

    @Override
    public int hashCode() {
        return Objects.hash(serial, date, machine, no);
    }
}

2. 处理Topic A的流：构造关联键并保留原键 #

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.JoinWindows;
import java.time.Duration;

public class KafkaJoinExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        ObjectMapper mapper = new ObjectMapper();

        // 读取Topic A
        KStream<String, String> streamA = builder.stream("topic-a", Consumed.with(Serdes.String(), Serdes.String()));

        // 转换流：生成关联键，同时保留原Topic A的键和frequency字段
        KStream<JoinKey, KeyValue<String, Integer>> transformedA = streamA.map((originalKey, value) -> {
            try {
                // 解析原键的JSON
                JsonNode keyNode = mapper.readTree(originalKey);
                Integer serial = keyNode.get("serial").asInt();
                String date = keyNode.get("date").asText();
                String machine = keyNode.get("machine").asText();

                // 解析value的JSON
                JsonNode valueNode = mapper.readTree(value);
                Integer no = valueNode.get("no").asInt();
                Integer frequency = valueNode.get("frequency").asInt();

                // 构造关联键
                JoinKey joinKey = new JoinKey(serial, date, machine, no);

                // 返回：关联键 -> (原A键, frequency)
                return KeyValue.pair(joinKey, KeyValue.pair(originalKey, frequency));
            } catch (Exception e) {
                // 处理解析异常，跳过错误记录
                return null;
            }
        }).filter((k, v) -> v != null); // 过滤掉解析失败的记录

        // 配置关联键的序列化器
        transformedA = transformedA.withKeySerde(Serdes.serdeFrom(
            new JsonSerializer<>(), new JsonDeserializer<>(JoinKey.class)
        ));

3. 处理Topic B的流：构造关联键并提取warning #

// 读取Topic B
        KStream<String, String> streamB = builder.stream("topic-b", Consumed.with(Serdes.String(), Serdes.String()));

        // 转换流：生成关联键，提取warning字段
        KStream<JoinKey, String> transformedB = streamB.map((originalKey, value) -> {
            try {
                JsonNode keyNode = mapper.readTree(originalKey);
                Integer serial = keyNode.get("serial").asInt();
                String date = keyNode.get("date").asText();
                String machine = keyNode.get("machine").asText();

                JsonNode valueNode = mapper.readTree(value);
                Integer no = valueNode.get("no").asInt();
                String warning = valueNode.get("warning").asText();

                JoinKey joinKey = new JoinKey(serial, date, machine, no);
                return KeyValue.pair(joinKey, warning);
            } catch (Exception e) {
                return null;
            }
        }).filter((k, v) -> v != null);

        transformedB = transformedB.withKeySerde(Serdes.serdeFrom(
            new JsonSerializer<>(), new JsonDeserializer<>(JoinKey.class)
        ));

4. 执行Inner Join并输出到Topic C #

// 执行inner join，设置窗口时间（根据业务调整，避免数据延迟漏匹配）
        KStream<JoinKey, KeyValue<KeyValue<String, Integer>, String>> joinedStream = transformedA.innerJoin(
            transformedB,
            (aValue, bValue) -> KeyValue.pair(aValue, bValue),
            JoinWindows.of(Duration.ofMinutes(5))
        );

        // 恢复原Topic A的键，构造输出value并发送到Topic C
        joinedStream.map((joinKey, combinedValue) -> {
            String originalAKey = combinedValue.key.key;
            Integer frequency = combinedValue.key.value;
            String warning = combinedValue.value;

            // 构造输出的JSON value
            String outputValue = String.format(
                "{\"no\": %d, \"frequency\": %d, \"warning\": \"%s\"}",
                joinKey.getNo(), frequency, warning
            );

            return KeyValue.pair(originalAKey, outputValue);
        }).to("topic-c", Produced.with(Serdes.String(), Serdes.String()));

        // 构建并启动流应用（省略配置部分）
        // KafkaStreams streams = new KafkaStreams(builder.build(), config);
        // streams.start();
    }
}

三、KSQL实现（更适合新手） #

如果不想写代码，可以用KSQL完成关联，步骤更直观：

-- 1. 创建Topic A的流，定义所有字段
CREATE STREAM stream_a (
    original_key STRING KEY,
    serial INT,
    date STRING,
    row INT,
    machine STRING,
    no INT,
    frequency INT
) WITH (
    KAFKA_TOPIC='topic-a',
    VALUE_FORMAT='JSON',
    KEY_FORMAT='JSON'
);

-- 2. 创建Topic B的流
CREATE STREAM stream_b (
    original_key STRING KEY,
    serial INT,
    date STRING,
    row INT,
    machine STRING,
    no INT,
    warning STRING
) WITH (
    KAFKA_TOPIC='topic-b',
    VALUE_FORMAT='JSON',
    KEY_FORMAT='JSON'
);

-- 3. 重新键化两个流，用关联字段作为新分区键
CREATE STREAM stream_a_rekeyed AS
SELECT original_key, serial, date, machine, no, frequency
FROM stream_a
PARTITION BY serial, date, machine, no;

CREATE STREAM stream_b_rekeyed AS
SELECT serial, date, machine, no, warning
FROM stream_b
PARTITION BY serial, date, machine, no;

-- 4. 执行关联并输出到Topic C
CREATE STREAM topic_c AS
SELECT a.original_key AS KEY,
       a.no,
       a.frequency,
       b.warning
FROM stream_a_rekeyed a
INNER JOIN stream_b_rekeyed b
WITHIN 5 MINUTES
ON a.serial = b.serial
AND a.date = b.date
AND a.machine = b.machine
AND a.no = b.no
EMIT CHANGES;

关键注意事项 #

• 复合键的equals/hashCode：必须正确实现，否则Kafka无法识别相同的关联键。
• 窗口时间：join是基于窗口的，要根据业务数据的延迟情况设置合理的窗口长度，避免因为数据到达先后导致匹配失败。
• 异常处理：解析JSON时可能出现格式错误，要添加异常处理逻辑，避免整个流崩溃。

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