如何使用kafka streams连接操作处理一对多关系

Question

你能帮助我如何使用Kafka流实现这一点吗？

场景：将订单数据的所有发票分组。在实时流媒体中，可能会延迟接收发票。因此，我们希望在加入之前等待20分钟将所有发票分组。

示例：订单“x”有3张发票，预计将在20分钟内收到。

预期输出：订单和3张发票应作为输出主题中的单个数据提供。

我们有下面的拓扑结构来实现这一点。

>

我们正在根据订单密钥对发票进行分组。我们设置了20分钟的翻滚窗口

将订单数据与生成的发票组关联

将输出写入新主题

问题：步骤3未等待步骤2完成。加入在收到订单后立即执行的操作。因此，我们没有得到预期的产出。

我们尝试使用joinwindows实现同样的功能。但是，由于联接窗口是滑动窗口，因此我们在输出主题中获得了重复的数据。

对于上面的示例，如果我们使用联接窗口而不是滚动窗口，我们将获得3个输出数据，订单分别有1张发票、2张发票和3张发票。

请帮助我解决此问题或建议任何替代方法

代码段：

 KTable<Windowed<String>, List<InvoiceList>> invoiceList= invoiceStream
                .groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofSeconds(1200)))
                .aggregate(() -> new ArrayList<InvoiceList>(),
                        (key, newValue, agg) -> {
                            new KeyValue<>(key, agg.add(newValue));
                            return agg;
                        },
                        Materialized.as("invoice-list").with(Serdes.String(), new ArrayListSerde<InvoiceList>(AppSerdes.InvoiceList())))
 
KStream<String, Order> orderOutput=
 
                orderStream.join(invoiceList, Joiner);
 
       
        orderOutput.to(AppConfig.OutputTopic.OUTPUT_ORDER,Produced.with(Serdes.String(), AppSerdes.Order()));

Answer 1

这种结合在我们的情况下起作用。因此，我们将其作为两个独立的流接收，并在消费者上添加了自定义逻辑来处理我们的用例。

谢谢

Answer 2

我想，订单先到，然后是发票，而不是相反。如果我的假设是正确的，那么你的逻辑就行不通了。因为当订单进入您的KStream时，可能没有发票，因此连接不会获取任何发票。请记住，KStream KTable连接是非窗口连接，可以像查找KTable（changelog流）一样使用。