Kafka-Streams - 在加入之前过滤 GlobalKTable

Question

您能否就以下问题的解决方法给我一个建议。我有两个主题，一个是静态内容，第二个是数据流。任务是加入数据，这在正常情况下很容易。我会将静态内容读取为 GlobalKTable，将动态内容读取为 KStream，然后简单地加入它们。问题是查找数据存在于同一主题的多个版本中。 “版本”由字段“validFrom”标识。因此流的数据需要根据其时间戳与对应版本的查找数据相结合。有没有办法过滤 GlobalKTable 中的数据？

最好的问候 马丁

Answer 1

您不能对GlobalKTable 本身应用过滤操作，但您可以尝试测试ValueJoiner 中记录的版本，并将未通过测试的连接结果记录的值设置为null。加入后，您可以应用过滤器过滤掉所有值为null 的记录。