apache-flink - ペアで以前に一致するイベントが見つからない場合、Flink はイベントを発行します

Question

2 つのイベント ストリームがあります。1 つはアイテムの有効期間の開始を通知するイベントを発行するストリームで、もう 1 つはアイテムの有効期間の終了を通知するイベントを発行するストリームです。(ストリームは で結合できますitemId。)

「寿命の終わり」イベントのみitemId1を持ち、対応する始まりを持たないそれぞれに対して、Flink で新しいイベントを発行するにはどうすればよいですか? (これらの開始イベントと終了イベントは、数時間または数日離れている可能性があります。)

Answer 1

でステートフルFlatMapFunctionを使用して機能を実装できますKeyedStream。

次のコード スニペットは、探していることのほとんどを行う必要があります。

val stream1: DataStream[Event1] = ???
val stream2: DataStream[Event2] = ???

// map both streams to their ID and a isStart flag to have a common type
val ids1: DataStream[(Int, Boolean)] = stream1.map(e => (e.id, true) )
val ids2: DataStream[(Int, Boolean)] = stream2.map(e => (e.id, false) )

// union both streams
val ids = ids1.union(ids2)

// use a stateful FlatMapFunction to check 
val onlyEOL: DataStream[Int] = ids
  // organize stream by ID
  .keyBy(_._1)
  // use stateful FlatMapFunction to check that bol arrived before eol
  .flatMapWithState { 
    (value: (Int, Boolean), state: Option[Boolean]) =>
      if (value._2) {
        // bol event -> emit nothing and set state to true
        ( List(), Some(true))
      } else {
        // eol event
        if (state.isDefined && state.get) {
          // bol was seen before -> emit nothing and remove state
          ( List(), None) 
        } else {
          // bol was NOT seen before -> emit ID and remove state
          ( List(value._1), None)   
        }
      }
  }