ストリームをピボット/アンピボットする方法

Question

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 
... 

と入力ストリームを持っている場合、私はあなたが出力タイプListで集約を使用することができ、出力ストリーム

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]] 
... 

Answer 1

を取得する方法を求めています。

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

Initializer戻り、空List<V>とAggregatorのようなものは、リストに値を追加します。

詳細は、ドキュメントと例をチェックアウト：

• https://docs.confluent.io/current/streams/developer-guide.html#aggregating
• https://github.com/confluentinc/kafka-streams-examples