在Kafka中执行批量验证并发送到相应的主题

Question

我在 Kafka 主题中存储了以下批处理格式：

Data generated -->  B2E, T3, T2, T1, B2S | B1E, T3, T2, T1, B1S  --> Data Consumed

这里BS表示批次开始，BE表示批次结束，t1,t2,t3是三个不同的数据。

这是扫描仪：

1. 在将批次结束记录推送到B1E主题之前，我们经过一些验证后得知该B1批次无效。

2. 在这种情况下，B1批次范围内BS的所有数据都BE应该转到特定主题。

因此，在上面的示例中，b1批次应转到主题 T1，b2批次应转到主题 T2。

我怎样才能使用卡夫卡做到这一点？

Answer 1

根据上图，按消息读取并验证每条消息，并更新内存窗口（带有批次数据的内存映射）中的一些消息，直到BE接收到相关批次结束。收到批次结束后，BE从窗口读取相关批次数据，并将所有批次记录发布到从处理（验证）结果中选择的主题。

对于窗口化，您可以使用内存映射，例如 <Key = BatchID> <Values = Array of Batch Data Object> 或 Kafka 流状态存储。

如果您需要类似 KStream 的解决方案，它将是如下所示的流