Kafka Streams - 時間ウィンドウに頻繁に表示されるメッセージのフィルタリング

Question

与えられた（ホッピング）時間ウィンドウの長さがTの場合に、しきい値がNよりも頻繁に表示されるメッセージをフィルタリングしようとしています。

例えば、以下の流れで上記簡略化のみになり、可能でない場合

#time, key 
0, A 
1, B 
2, A 
3, C 
4, D 
5, A 
6, B 
7, C 
8, C 
9, D 
10, A 
11, D 
12, D 
13, D 
14, D 
15, D 

とN=2とT=3、結果は、代替的

0, A 
2, A 
7, C 
8, C 
9, D 
11, D 
12, D 
13, D 
14, D 
15, D 

であるべきではしきい値が満たされた後のメッセージのフィルタリング：

#time, key 
2, A 
8, C 
11, D 
12, D 
13, D 
14, D 
15, D 

カフカストリームでこれが可能ですか？

これまでストリームのwindowed count（インスタンスはKTable）を作成し、元のストリームに戻してみました。 windowed countのキーをKTable#toStream((k,v) -> k.key())を使用してdummy aggregationのインスタンスをKTableに戻して元のキーに戻します。これは、leftJoinにスレッシュホールドを超過した後に非常に近いメッセージを見逃す遅延を引き起こすようです。

source.join(wcount, (leftValue, rightValue) -> rightValue, JoinWindows.of(TimeUnit.SECONDS.toMillis(5)),stringSerde, stringSerde, longSerde) 
      .filter((k,v) -> v!=null) 
      .filter((k,v) -> v>1) 
      .print("output"); 

wcountに各UPSERTがイベントをトリガーするので、これは、重複出力をもたらす：KStream（ダミー集合を除外）適切なウィンドウと結合 - 私もKStreamを実行しようとした

final Serde<String> stringSerde = Serdes.String(); 
    final Serde<Long> longSerde = Serdes.Long(); 

    KStream<String, Long> wcount = source.groupByKey() 
      .count(TimeWindows.of(TimeUnit.SECONDS.toMillis(5)),"Counts") 
      .toStream((k,v) -> k.key()); 

    // perform dummy aggregation to get KTable 
    KTable<String, Long> wcountTable = wcount.groupByKey(stringSerde, longSerde) 
       .reduce((aggValue, newValue) -> newValue, 
       "dummy-aggregation-store"); 

    // left join and filter with threshold N=1 
    source.leftJoin(wcountTable, (leftValue, rightValue) -> rightValue,stringSerde, stringSerde) 
      .filter((k,v) -> v!=null) 
      .filter((k,v) -> v>1) 
      .print("output"); 

。

Answer 1

。リスト内のすべての生データを収集するウィンドウ集約を適用することができます（つまり、ウィンドウを手動で実現します）。その後、ウィンドウを評価するflatMapを適用します。しきい値がまだ満たされていない場合は、何も出さない。しきい値が初めて満たされた場合は、すべてバッファされたデータを送信します。フラットマップの値がしきい値より大きい場合は、リスト内の最新のものだけを出力します（以前はflatMapの呼び出しではなく、新しく追加されたものだけを出力していました）。

注：KTableキャッシュを無効にする必要があり、すなわち、それ以外の場合は、アルゴリズムが正しく動作しませんconfigパラメータ「cache.max.bytes.buffering」= 0を設定します。

このような何か：

KStream<Windowed<K>, List<V>> windows = stream.groupByKey() 
               .aggregate(
               /*init with empty list*/, 
               /*add value to list in agg*/, 
               TimeWindows.of()...), 
               ...) 
               .toStream(); 
KStream<K,V> thresholdMetStream = windows.flatMap(
              /* if List#size < threshold 
               then return empty-list, ie, nothing 
               elseif List#size == threshold 
               then return whole list 
               else [List#size > threshold] 
               then return last element from list 
              */); 

Answer 2

AFAIKこれはCount-Min-Sketchアルゴリズムに最適です。例えば、ストリーム-libの実装を参照してください：これは確かに可能である

https://github.com/addthis/stream-lib