如何对比两个Kafka Table并将Table1中不存在的数据发送至指定Topic

现有配置与代码 #

Kafka Streams配置类 #

@Configuration
@EnableKafkaStreams
@EnableKafka
public class KafkaConfig {

  @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
  KafkaStreamsConfiguration kstreamConfig(){
    Map<String,Object> props=new HashMap<>();
    props.put(APPLICATION_ID_CONFIG,"MY_GROUP_ID");
    props.put(BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
    props.put(DEFAULT_KEY_SERDE_CLASS_CONFIG,Serdes.String().getClass().getName());
    props.put(DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    return new KafkaStreamsConfiguration(props);
  }
}

现有业务代码 #

KTable<String,String> table1=streamsBuilder.table("TOPIC1");
KTable<String,String> table2=streamsBuilder.table("TOPIC2");

实现方案 #

要提取Table1中存在但Table2中不存在的数据，可通过左外连接+过滤的方式实现：

1. 对两个KTable执行左外连接，保留Table1的所有键值对，若Table2中无对应键，则对应值为null
2. 过滤出Table2值为null的记录，即为Table1独有的数据
3. 将过滤后的结果发送到目标Topic（示例中命名为TOPIC_ONLY_IN_TABLE1）

完整业务逻辑代码如下：

// 左外连接：保留table1的所有记录，匹配table2中存在的记录，不存在则table2Value为null
KTable<String, String> table1Only = table1.leftJoin(table2,
    (table1Value, table2Value) -> table2Value == null ? table1Value : null)
// 过滤掉table2Value不为null的记录（即过滤掉两个表都存在的记录）
.filter((key, value) -> value != null);

// 将结果发送到目标Topic
table1Only.toStream().to("TOPIC_ONLY_IN_TABLE1");

关键说明 #

• 左外连接（leftJoin）是核心：左表（table1）的所有记录都会被保留，右表（table2）仅保留匹配到的记录，未匹配的则为null
• 过滤操作只保留右表值为null的记录，确保最终结果是table1独有的数据
• 由于KTable是基于键的状态存储，该逻辑会自动处理数据的更新（比如table2新增了某个键后，后续该键会被过滤掉）

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