如何对 KTable (Kafka) 中的值求和答案

【问题标题】：How to sum the values in a KTable (Kafka)如何对 KTable (Kafka) 中的值求和
【发布时间】：2020-07-30 17:12:45
【问题描述】：

我有两个 KTable。一个持有我持有的股票数量，另一个持有这些股票的最新价格。

例如，

Quantities
Key    Value
AAPL   50
TSLA   100

Prices
Key    Value
AAPL   10
TSLA   20

当 either 表有更新时，我想计算我的投资组合的总价值并将其发布到另一个主题。在上面的示例中，总值将为50 x $10 + 100 x $20 = $2,500。

最好的方法是什么？我想一个明智的第一步是加入表格，并分别计算每个位置的值。但我不知道如何将生成的 KTable 中的值相加以获得我的投资组合的总价值？

【问题讨论】：

【解决方案1】：

建议是加入，按键分组，然后映射产品

形成一个流

AAPL (50, 10)
TSLA (10, 20)

然后mapValues得到

AAPL 500 
TSLA 200

一旦你有了 KTable，你就可以遍历 KVStore 来对总值求和

【讨论】：

谢谢。如何定期迭代 KVStore？或者，最好是每次数量或价格发生变化？
您可以使用 ScheduledExecutor 进行定期更新，也可以使用单独的生产者/消费者对作为通知渠道
谢谢。我很惊讶 DSL 或处理器 API 没有更简单的方法。对 KTable 的变化值求和似乎不是一项艰巨的任务..
该表是一个表，而不是一个单一的值 :) 非常欢迎您制作一个辅助流，对所有值执行 reduce，从而得出一个总和跨度>
作为最后一步，您可以执行 KTable.groupBy().aggregat() 并将所有记录映射到同一个键并在您的 aggregation() 中应用总和 - 请注意，这将是一个非并行操作并且不会扩展。