如何旋转/取消旋转流答案

【问题标题】：How to pivot / un pivot a stream如何旋转/取消旋转流
【发布时间】：2017-10-20 08:00:33
【问题描述】：

有一种方法可以使用 Kafka 流来 PIVOT / UNPIVOT（爆炸、转移）流吗？

如果我有输入流

machineId   ts  VarName VarValue
m1  2017-10-01 00:00:00 var1    1.0
m1  2017-10-01 00:00:00 var2    2.0
m2  2017-10-01 00:00:00 var1    3.0
m2  2017-10-01 00:00:00 var3    4.0
m3  2017-10-01 00:00:00 var4    5.0
...

我想要一种获取输出流的方法

machineId   ts  Vars
m1  2017-10-01 00:00:00 [[var1, 1.0], [var2, 2.0]]
m2  2017-10-01 00:00:00 [[var1, 3.0], [var3, 4.0]]
m3  2017-10-01 00:00:00 [[var4, 5.0]]
...

【问题讨论】：

标签： apache-kafka-streams

【解决方案1】：

您可以使用输出类型为List 的聚合。类似的东西

KStream<MachineId, V> inputStream = ...
KTable<MachineId, List<V>> result = inputStream.groupByKey()
                                               .aggregate(...);

Initializer 返回一个空的List<V>，Aggregator 会将值附加到列表中。

查看文档和示例了解更多详情：

【讨论】：