Realtime Compute for Apache Flink:カスタムトリガーの開発

トリガーの機能

1. ウィンドウに入るデータ要素を監視します。

2. 時間やイベント数など、事前定義された条件に基づいて処理をトリガーするかどうかを決定します。

3. 次の操作のいずれかを返します。

   • CONTINUE: データの収集を続けます。

   • FIRE: 計算をトリガーし、状態を保持します。

   • FIRE_AND_PURGE: 計算をトリガーし、状態をクリアします。

   • PURGE: 計算をトリガーせずに状態をクリアします。

組み込みトリガー

Flink は、時間ベースまたはカウントベースのウィンドウ操作向けに、次の組み込みトリガーを提供しています。

実際のユースケースでは、多くの場合、以下を行う必要があります。

• 5 つのイベントまたは 1 分後にウィンドウをトリガーするなど、複数のトリガー条件を使用します。

• ユーザーログアウトや注文完了など、特定のイベントに基づいてウィンドウをトリガーします。

• 早期起動、遅延起動、複数起動など、ウィンドウのライフサイクルを管理します。

• 遅延データによってウィンドウが再度トリガーされるなど、デフォルトの動作の副作用を回避します。

これらの複雑な要件は、組み込みトリガーでは満たすことができないため、カスタムトリガーが必要になります。

1 時間のタンブリングイベントタイムウィンドウでは、ウィンドウが終了するとウィンドウ計算がトリガーされます。 さらに、そのウィンドウ内でユーザーの 5 番目のイベントが到着すると、即時計算がトリガーされます。 この早期トリガーは、ウィンドウごとに最大 1 回起動します。

サンプルコード:

public class CustomCountTrigger extends Trigger<UserEvent, TimeWindow> {

    // ウィンドウ内の各キーの要素数を記録します
    private final ValueStateDescriptor<Integer> countStateDesc =
            new ValueStateDescriptor<>("count", Integer.class);
    // 計算がすでにトリガーされているかどうかを記録します
    private final ValueStateDescriptor<Boolean> flagStateDesc =
            new ValueStateDescriptor<>("flag", Boolean.class);

    // 新しい要素がウィンドウに入ると onElement() を呼び出します
    @Override
    public TriggerResult onElement(UserEvent event, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
        // 現在のウィンドウ内の現在のキーのカウント状態を取得します
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        ValueState<Boolean> flagState = ctx.getPartitionedState(flagStateDesc);
        int count = countState.value() == null ? 0 : countState.value();
        boolean flag = flagState.value() == null ? false : flagState.value();

        // 要素ごとにカウントを増分します
        count += 1;
        countState.update(count); // 状態を更新します

        // カウントが 5 に達すると、ウィンドウ計算をすぐにトリガーします
        if (count >= 5 && !flag) {
            flagState.update(true); // この追加計算が 1 回だけトリガーされるように状態を更新します
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext triggerContext) throws Exception {
        // このトリガーでは、処理時間タイマーをスキップします。
        return TriggerResult.CONTINUE;
    }

    // 登録済みのイベントタイムタイマーが起動すると onEventTime() を呼び出します（例: ウィンドウの終了時刻）
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE; // トリガーが起動し、ウィンドウがクリアされます
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext ctx) throws Exception {
        // ウィンドウの状態をクリーンアップします
        ctx.getPartitionedState(countStateDesc).clear();
        ctx.getPartitionedState(flagStateDesc).clear();
    }
}

ユースケース 1

アラートのために、ウィンドウが終了する前に複数回計算をトリガーします。

1. flagStateDesc を削除して、計算を複数回トリガーできるようにします。

2. または、カウントマーカーを追加して、特定の回数のトリガー起動（アラートイベントなど）の後でのみウィンドウを終了します。

   ### flagStateDesc を削除した後、5 つのレコードが出力されます。
   キー: 101, アクセス数: 5       // count >= 5 のため、早期計算がトリガーされ、状態はクリアされません。
   キー: 101, アクセス数: 6       // count >= 5 のため、早期計算がトリガーされます。
   キー: 101, アクセス数: 7      // count >= 5 のため、早期計算がトリガーされます。
   キー: 101, アクセス数: 8      // count >= 5 のため、早期計算がトリガーされます。
   キー: 101, アクセス数: 8      // ウィンドウが終了すると計算がトリガーされ、状態がクリアされます。

3. トリガーは、計算をトリガーするタイミングのみを決定します。 実際のオブジェクト情報は .process から取得されます。ここでは、数量と条件の状態に基づいて異なる収集結果を生成するように指定できます。

ユースケース 2

ウィンドウを閉じるには、ウォーターマークを進める必要があります。 長期間データが生成されない場合、タイムリーなウィンドウ計算をどのように保証できますか?

• ソリューション 1: 処理時間ウィンドウを使用する

  イベント時間セマンティクス（つまり、イベントが発生した時刻は問題ではない）が不要な場合は、処理時間ウィンドウを使用できます。

  .keyBy(keySelector)
  .window(TumblingProcessingTimeWindows.of(Time.seconds(60))) // 処理時間を使用します
  .process(new MyProcessWindowFunction())

• ソリューション 2: withIdleness メソッドを使用する

  Flink の WatermarkStrategy は、withIdleness メソッドを提供して、指定された期間非アクティブになった後、データソースを自動的にアイドルとしてマークし、ウォーターマークの生成をブロックしないようにします。

  // アイドル状態のデータソースは、最小ウォーターマークの計算に関与しなくなり、アクティブなデータソースのウォーターマークの進行を妨げなくなります。
  WatermarkStrategy
      .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
      .withIdleness(Duration.ofMinutes(1));  //データソースまたはパーティションに 1 分間イベントがない場合、アイドル状態としてマークされることを示します

• ソリューション 3: カスタムウォーターマークジェネレーターを使用する

  イベント時間セマンティクスを使用する必要がある場合は、新しいデータが到着しなくてもウォーターマークが前進し続けるカスタムウォーターマークジェネレーターを定義します。

  • onEvent() メソッドで最新のイベントの到着時刻を記録します。

  • onPeriodicEmit() メソッドで、現在の時刻とイベントが最後に受信された時刻の間隔を確認します。

  • 間隔が設定されたしきい値を超える場合は、データソースをアイドル状態と見なし、ウォーターマークの生成をスキップするか、特定のウォーターマークを直接生成します。

  public class IdleAwareWatermarkGenerator implements WatermarkGenerator<MyEvent> {
      private long lastEventTimestamp = Long.MIN_VALUE;
      private final long maxIdleTimeMs; // 最大アイドル時間
  
      public IdleAwareWatermarkGenerator(long maxIdleTimeMs) {
          this.maxIdleTimeMs = maxIdleTimeMs;
      }
  
      @Override
      public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
          lastEventTimestamp = Math.max(lastEventTimestamp, eventTimestamp);
      }
  
      @Override
      public void onPeriodicEmit(WatermarkOutput output) {
          long currentTime = System.currentTimeMillis();
          if (lastEventTimestamp == Long.MIN_VALUE || currentTime - lastEventTimestamp > maxIdleTimeMs) {
              // 長時間イベントが到着しない場合は、新しいウォーターマークを発行しません
              return;
          }
          output.emitWatermark(new Watermark(lastEventTimestamp));
      }
  }

  WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
      .forGenerator((ctx) -> new IdleAwareWatermarkGenerator(60_000)) // 最大アイドル時間を 60 秒に設定します
      .withTimestampAssigner((event, timestamp) -> event.getEventTime());

• ソリューション 4: タイマーを登録する

  @Override
  public TriggerResult onElement(Event event, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
      ctx.registerEventTimeTimer(window.maxTimestamp()); // イベントタイムタイマーを登録します
      ctx.registerProcessingTimeTimer(window.maxTimestamp() + 1000); // 耐障害性: イベントがなくてもタイマーがトリガーされるようにします
      return TriggerResult.CONTINUE;
  }
  
  @Override
  public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
      return TriggerResult.FIRE_AND_PURGE; // 強制的なトリガーとクリーンアップ
  }

• ソリューション 5: 外部システムからハートビートメッセージを送信する

  Kafka または ワークフロー を使用して、定期的なハートビートメッセージをダウンストリームに送信し、ウィンドウのクローズをトリガーします。

サマリー