聚合用例-按两个不同的kafka流分组和计算平均值

Question

我被困在一个用例中。我需要计算花费在娱乐上的百分比。

在我的流中，我得到了具有行业代码和花费金额的记录(行业代码是基于娱乐的，而不是基于娱乐的)

e.g>娱乐行业代码>行业代码量

                                                         157       100

                                                         257        200

                                                         157         300

非娱乐行业代码>行业代码金额

                                                         457       100

                                                          657       200

                                                          457       300

所以我需要计算花在娱乐上的百分比应该是娱乐金额的总和/(非娱乐金额的娱乐amount+total的总和)

我尝试的解决方案是:创建两个Ktables，一个娱乐分组，即157400( 100+300)

                                   257    100

另一款非娱乐KTable 457 400

                                 657      200

我现在有两个KTables，但是我如何计算%呢？这种方法正确吗？

Answer 1

到目前为止，我不完全确定你到底尝试了什么。当您创建两个表时，两个表是否都有多行(即每个代码一行)，或者您已经将不同的代码“合并”在一起了吗？如果每个表包含多行，则需要通过设置组成的代理键(如所有行的整数值0)将所有行聚合在一起：

KStream sumSpendingEntertainment = spendingEntertainment.groupBy((k,v) -> 0)
                                                        .aggregate(...);
KStream sumSpendingAll = spendingAll.groupBy((k,v) -> 0)
                                    .aggregate(...);

最后，您得到两个包含一行的KTable；一个包含总支出，另一个包含“娱乐支出”，两个KTables将使用相同的组成代理键(在我们的示例中为整数0)。

在最后一步中，您可以连接两个表，以将两个总和相除：

sumSpendingEntertainment.join(sumSpendingAll,
                              (sumEnt, sumAll) -> sumEnt / sumAll); // this is the `ValueJoiner` expressed as lambda