在 KStream 和 KTables 上同时操作

Question

我正试图在Kafka Streams中实现一个用例，在该用例中，我基于对该流应用一些过滤器来填充Ktable，让我们将该表称为跟踪表，其中键是从事件派生的，值是事件。现在，对于后续事件，我检查此表以验证它们是否被跟踪，并更新事件(如果被跟踪(或将其发布到其他主题。我不知道如何同时做到这一点。这是我迄今为止所拥有的。

// Branch based on conditions
KStream<String, Event>[] segregatedRecords = branches[0]
.branch((key, event) -> event.getStatus().getStatus().equals("A"),
(key, event) -> event.getStatus().getStatus().equals("B"),
(key, event) -> event.getStatus().getStatus().equals("C"),

// Store events with status A to a topic
segregatedRecords[0]
.selectKey((key, event) -> createKey(event))
.mapValues(transform)
.to(intermediateTopic);
// Load topic from previous step as GlobalKtable
GlobalKTable<String, Event> trackedEvents = streamsBuilder.globalTable(intermediateTopic);
// The following step is where I'm stuck, because I can not perform conditional actions
// If the event exists in the tracking table (update) but if not then how to publish it to a different topic?
segregatedRecords[1]
// derive key for lookup
.selectKey((key, event) -> createKey(event))
// update the event status in the table 
.join(trackedEvents, (key, value) -> key,(event, tracked) -> modifiedEvent
).to(intermediateTopic);
// Other events will need to refer the latest information in the tracked table for further processing 

Answer 1

您可以通过将segregatedRecords[1]分支为2个子Topology来实现这一点，一个分支执行表锁定作为您的代码，另一个分支使用低级处理器API(在本例中使用transformValues(来检查底层GlobalKTable状态存储是否包含新派生密钥的记录，如果该记录存在，则将Event转换为nullEvent，然后我们过滤掉Event为空的事件(因为我们已经在您的第一个子拓扑中加入了这些事件(。我更新了你的代码一点点：

//give your GlobalKTable a name to query later
GlobalKTable<String, Event> trackedEvents = streamsBuilder.globalTable(intermediateTopic, Materialized.as("tracked_event_global_store"));
KStream<String, Event> derivedKStream = segregatedRecords[1]
// derive key for lookup
.selectKey((key, event) -> createKey(event));
// this sub-topology perform table lockup as normal: update the event status in the table
derivedKStream.join(trackedEvents, (key, value) -> key,(event, tracked) -> modifiedEvent)
.to(intermediateTopic);
// this sub-topology check whether the event existed in trackedEvents, if yes then event has been already joined 
// so we transform to null value and filter in next step 
derivedKStream.transformValues(() -> new ValueTransformerWithKey<String, Event, Event>() {
//get the underlying store of Tracked GlobalKTable
KeyValueStore<String, Event> trackedKvStore;
@Override
public void init(ProcessorContext context) {
//using the previous name
trackedKvStore = (KeyValueStore<String, Event>) context.getStateStore("tracked_event_global_store");
}
@Override
public Event transform(String derivedKey, Event event) {
//if event existed in trackedEvents then return a null event so we can filter out in next pipe
if (trackedKvStore.get(derivedKey) != null) {
return null;
}
//event not exist in trackedEvents, keep the event and send to different topic
return event;
}
@Override
public void close() {
}
})
.filter((derivedKey, event) -> event != null)
.to("your different toic name");

更新：关于不能从单个主题intermediate创建GlobalKTable和KStream的问题(不能像这里描述的那样多次读取主题(：

1. 为GlobalKTable创建一个专用输入主题(此主题必须启用日志压缩(：

KStream<Object, Object> intermediateKStream = streamsBuilder.stream(intermediate);
intermediateKStream.to(trackedInputTopic);
//instead of building GlobalKTable from intermediate, use this dedicated topic trackedInputTopic
GlobalKTable<String, Event> trackedEvents = streamsBuilder.globalTable(trackedInputTopic, Materialized.as("tracked_event_global_store"));
//Perform things you want to do with the intermediate topic
intermediateKStream
...