Realtime Compute for Apache Flink:Flink を使用したリアルタイムデータ集約

背景と課題

従来のリアルタイムデータ集約システムは、実際のユースケースにおいて、以下のような複数の課題に直面します。

• 遅延データの処理: 分散リアルタイム処理では、ネットワーク遅延、システムジッター、またはアップストリームの変動により、データが順序通りに到着しないことがよくあります。ウォーターマーク、ステートロールバック、またはウィンドウベースの再処理などの適切な遅延データ処理メカニズムがないと、既存データが誤った結果で上書きされる可能性があります。これにより、統計的な偏りが生じ、モニタリングの精度や意思決定の信頼性に影響を与えます。

• 複雑なステート管理: 従来、高次元データやデータスキューを処理すると、ステートサイズが指数関数的に増大する可能性があります。大きなステートサイズは、過剰なメモリリソースを消費し、チェックポイント作成を遅らせ、回復時間を増加させ、システムの安定性を損ない、ジョブの失敗リスクを高めます。

• リソース消費とパフォーマンスのバランス: リアルタイム集約では、計算オーバーヘッド、ストレージ使用量、結果の精度など、いくつかの要因をトレードオフする必要があります。インメモリステートに過度に依存するとリソースコストが増加し、外部ストレージへの頻繁な読み書きは I/O ボトルネックを生み出し、スループットと待機時間に影響を与える可能性があります。

ソリューションの比較

従来のステートフル集約

このソリューションは、Flink の組み込みメカニズムを使用して、リアルタイムデータ集約のためにメモリ内にステートを維持します。

以下の SQL スニペットは、ログストリーム (view_source) をリアルタイムで集約します。時間 (ts) とクラスターごとにページビュー (pv) とクリック数を計算し、結果を sink_table に書き込みます。

INSERT INTO sink_table
SELECT 
  ts, cluster,
  SUM(pv) as pv,
  SUM(click) as click 
FROM view_source 
GROUP BY ts, cluster;

ソリューションの説明

これは、リアルタイム集約の基本的なソリューションです。Flink のステートフル集約機能を使用し、使い慣れた SQL 構文を使用してデータ集約タスクを開発できます。データがオペレーターを通過すると、エンジンは事前定義されたキーに基づいてステートを検索して更新し、集約された結果をダウンストリームストレージに書き込みます。

利点と制限

このソリューションは実装がシンプルで、高いデータ鮮度を提供します。しかし、明確な制限があります。

• ステートの有効期限が切れた後の遅延データによる集約の破損: 最終結果に正しく集約されたタイムウィンドウ (例: pv=999) を考えてみましょう。このウィンドウのステートが (例えば 09:30 に) 期限切れになると、そのウィンドウ内のタイムスタンプを持つ後続の遅延イベントは既存のステートを見つけられません。代わりに、Flink は新しいステートを初期化し、不完全な結果 (例: pv=1) を出力し、それがシンク内の以前の正しい結果を上書きします。

• ステートの肥大化によるパフォーマンスの低下: 大規模なデータセットでの高い同時実行性は、集約ステートのホットスポットを引き起こす可能性があります。これにより、ステートが急速に拡大し、メモリ消費量が増加し、チェックポイント作成が大幅に遅くなり、システムの安定性と回復時間に影響を与えます。

• 長引く障害回復: システムの障害回復中、ステート全体をチェックポイントからロードする必要があります。ステートが大きければ大きいほど、回復時間が長くなり、システムの可用性に影響を与えます。

UDAF を使用したステートレス増分集約

このソリューションは、複雑なステート管理をストレージシステムにプッシュダウンします。ストリーム処理エンジンはステートレスな増分計算を実行し、最終的な集約結果の計算をダウンストリームストレージに依存します。

以下のユーザー定義の集計関数 (UDAF) は、ステートレス集約を実装します。現在のバッチまたはウィンドウ内のデータに対して増分計算を実行します。

public class LongSumAggUDAF extends AggregateFunction<Long, LongAccumulator> {
    @Override
    public LongAccumulator createAccumulator() {
        return new LongAccumulator();
    }
    
    public void accumulate(LongAccumulator acc, Long value) {
        acc.add(value); // 現在のマイクロバッチ内のデータを集約します
    }
    
    @Override
    public Long getValue(LongAccumulator acc) {
        return acc.getValue();
    }
}

ソリューションの説明

• UDAF: 標準の集計関数とは異なり、UDAF はオペレーターレベルで現在のバッチまたはウィンドウ内のデータに対してのみ増分計算を実行します。バッチ間で履歴ステートを維持せず、現在のマイクロバッチの値のみに焦点を当てます。

• 「読み取り-計算-書き込み」による増分集約: シンクコネクタは、まずストレージから特定のプライマリキーに対する最後に知られている集約値を読み取ります。次に、履歴値と現在のデータバッチからの増分結果をマージします。最後に、新しい集約結果をストレージに書き戻します。この「読み取り-計算-書き込み」パターンは、ジョブが再起動したり遅延データが到着したりした場合でも、正しい集約を保証します。

利点と制限

このソリューションは遅延データを完璧に処理します。ステートの有効期限切れによる結果の誤った上書きを防ぎます。さらに、エンジンが大きなステートサイズを維持しなくなるため、メモリ効率とチェックポイントのパフォーマンスが大幅に向上します。しかし、欠点として、このソリューションは各バッチまたはウィンドウに対して追加の読み取り操作を必要とするため、ストレージシステムへの負荷と処理の待機時間が増加します。

中間集約テーブル

このソリューションでは、Paimon テーブルなどのデータレイクテーブルを導入して、集約結果を保存します。図に示すように、集約パイプラインは次のように機能します: Flink は生データに対してステートレス集約を実行し、レイクテーブルで結果をマージして、ダウンストリームに送信します。

以下のコードは Paimon テーブルを作成します。プライマリキーが一致すると、pv と click カラムが sum を使用して集約されます。

-- Paimon 集約テーブルを作成します
CREATE TABLE paimon_agg (
    ts TIMESTAMP(3),
    cluster STRING,
    pv BIGINT,
    click BIGINT,
    PRIMARY KEY (ts, cluster) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.pv.aggregate-function' = 'sum',
    'fields.click.aggregate-function' = 'sum'
);

ソリューションの説明

このソリューションは、Paimon コネクタの データマージメカニズム を使用してデータを集約します。'merge-engine' = 'aggregation' を設定し、集計関数を指定すると、Paimon コネクタは特定のカラムの新しい結果と履歴結果を自動的にマージします。ワークフローには以下が含まれます。

1. データストリームの処理: ステートレス集約を実行し、結果を Paimon テーブルに書き込みます。

2. データのマージ: 集約結果をマージし、バージョンを管理します。

3. 結果の送信: 結果をバッチで、かつ定期的な間隔でダウンストリームに送信します。

利点

• ダウンストリームストレージへの負荷の軽減: バッチ同期により、リアルタイムの書き込み負荷が大幅に削減され、ダウンストリームストレージの安定性が向上します。

• 集約ロジックの簡素化: このパイプラインは、「読み取り-計算-書き込み」の必要性を排除します。データインジェスチョンと増分処理に焦点を当てています。

• 柔軟性とメンテナンス性の向上: 各コンポーネントのタスクが明確に定義されています。さらに、Paimon テーブルはさまざまなダウンストリームシステムとシームレスに統合されます。

• バージョン管理と履歴スナップショットの参照

ソリューションの選択と考慮事項

ソリューションの選択には、コスト、効率、複雑さ、データ規模、技術的能力の間のトレードオフが伴います。以下に、各ソリューションの利点と制限の概要を示します。

• ステートフル集約: 実装がシンプルで、高い鮮度を提供します。しかし、遅延データの処理とステートの効果的な管理に課題があります。

• ステートレス増分集約: 遅延データを効果的に処理し、O&M のオーバーヘッドが低く、広く適用可能です。ただし、集約ロジックに合わせた UDAF を定義する必要があります。

• 中間集約テーブル: ダウンストリームシステムへの負荷を軽減しながら、大規模なデータセットを処理できます。しかし、実装はより複雑で、かなりのセットアップとメンテナンスが必要です。

開発と O&M のオーバーヘッドは、多くの場合、逆相関の関係にあることに注意してください。ソリューションを選択する際には、コスト、効率、複雑さ、データ規模、技術的能力のバランスを取る必要があります。