Realtime Compute for Apache Flink:大規模ステート SQL ジョブをチューニングしてバックプレッシャーを削減する

適切な TTL 値の設定によるステートサイズの制御

Realtime Compute for Apache Flink の開発コンソールの [デプロイメント] ページで、状態の有効期限 パラメーターを設定します。詳細については、「パラメーター」をご参照ください。

TTL が短すぎると、計算結果が不正確になる可能性があります。例えば、データが遅れて到着し、集約や結合の際に関連するステートが期限切れになっている場合、結果は誤りとなります。TTL が長すぎると、リソース消費が増加し、安定性が低下します。データ特性とビジネス要件に基づいて TTL を設定してください。例えば、日次計算で日をまたぐ最大ドリフトが 1 時間の場合、TTL を 25 時間に設定します。

JOIN_STATE_TTL を使用したストリームごとの TTL (VVR 8.0.1 以降)

Ververica Runtime (VVR) 8.0.1 以降では、JOIN_STATE_TTL ヒントを使用して、通常結合の左ストリームと右ストリームに異なる TTL 値を設定できます。これにより、不要なステートストレージを削減し、パフォーマンスを向上させます。構文の詳細については、「クエリヒント」をご参照ください。

SELECT /*+ JOIN_STATE_TTL('left_table' = '..', 'right_table' = '..') */ *
FROM left_table [LEFT | RIGHT | INNER] JOIN right_table ON ...

次の表は、JOIN_STATE_TTL を実際のデプロイメントで使用した場合の影響を示しています。

項目	デプロイメント詳細	ステートサイズ
変更前	左ストリームのデータ量は右ストリームの 20 倍から 50 倍です。パフォーマンスを維持するため、右ストリームの TTL は 10 日に設定されていますが (本来は 18 日であるべき)、データの正確性が損なわれています。ステートサイズは約 5.8 TB です。デプロイメントは最大 700 CU を消費します。
その後	JOIN_STATE_TTL により、左ストリームの TTL は 12 時間、右ストリームの TTL は 18 日に設定され、データ整合性が確保されます。ステートサイズは約 590 GB に減少し、元の 10 分の 1 になります。デプロイメントは 200 から 300 CU を消費し、リソースを 50% から 70% 節約します。

不要なステートフルオペレーターの削除

この方法は、オプティマイザーによって派生するオペレーターにのみ適用されます。SQL によって呼び出されるオペレーターは、通常、クエリロジック自体によって要求されます。

ChangelogNormalize

このオペレーターは、イベント時間に基づくテンポラル結合を除き、アップサートソーステーブルを含むほとんどのシナリオで呼び出されます。Upsert Kafka コネクタまたは同様のコネクタを使用する前に、イベント時間に基づくテンポラル結合が実行されていないことを確認してください。実行時に ChangelogNormalize オペレーターのステート関連のメトリックを監視します。ソーステーブルに多数のプライマリキーがある場合、オペレーターがキー付きステートを維持するため、ステートサイズが増加します。頻繁なプライマリキーの更新も、ステートへのアクセスと変更の頻度を増加させます。

データ同期のシナリオで Upsert Kafka コネクタを使用することは避けてください。代わりに、exactly-once 処理を保証するデータ同期ツールを使用してください。

SinkUpsertMaterializer

デフォルトでは、table.exec.sink.upsert-materialize パラメーターは auto に設定されています。これは、順序が乱れたチェンジログレコードなど、特定のシナリオで正確性を保証するために、システムが自動的に SinkUpsertMaterializer オペレーターを追加することを意味します。このオペレーターが存在することが、必ずしもレコードの順序が乱れていることを意味するわけではありません。例えば、複数のキーでデータをグループ化し、それらを単一の列にマージする場合、オプティマイザーはアップサートキーを正確に導出できないため、安全策としてこのオペレーターを追加します。

データ分布パターンが十分に理解されており、このオペレーターがなくても最終結果が正しい場合は、table.exec.sink.upsert-materialize を none に設定してパフォーマンスを向上させることができます。

Realtime Compute for Apache Flink の開発コンソールで SinkUpsertMaterializer がアクティブかどうかを確認します。存在する場合、トポロジー図では Sink オペレーターと連結して表示されます。

SinkUpsertMaterializer オペレーターが使用されておらず、計算結果が正しい場合は、'table.exec.sink.upsert-materialize'='none' を追加して不要な使用を防ぎます。設定手順については、「デプロイメント実行中のパラメーターを設定する方法」をご参照ください。

Ververica Runtime (VVR) 8.0 以降は、SQL 実行計画のインテリジェントな分析をサポートし、このような問題の特定に役立ちます。

miniBatch によるステートアクセス頻度の削減

分単位のレイテンシが許容できる場合は、miniBatch を有効にしてステートへのアクセスと更新の頻度を減らします。詳細については、「miniBatch を有効にしてスループットを向上させる」をご参照ください。

Realtime Compute for Apache Flink では、以下のオペレーターが miniBatch をサポートしています。

実行計画の最適化

オプティマイザーは、SQL 文と構成に基づいてステートの実装を選択し、実行計画を生成します。以下の調整により、ステートサイズを大幅に削減できます。

通常結合でのプライマリキーの使用

• 結合キーにプライマリキーが含まれている場合、システムは ValueState<RowData> オブジェクトを使用して、結合キーごとに最新の値のみを格納します。これにより、ストレージ効率が最大化されます。

• 結合キーにプライマリキー以外のキーが含まれている場合、システムは MapState<RowData, RowData> オブジェクトを使用して、各結合キーのプライマリキーごとにソーステーブルの最新レコードを格納します。

• プライマリキーが定義されていない場合、システムは MapState<RowData, Integer> オブジェクトを使用して、各結合キーのデータレコード全体と出現回数を格納します。

DDL 文でプライマリキーを定義し、プライマリキーで通常結合を実行して、ストレージ効率を最適化してください。

追加専用ストリームでの重複排除に ROW_NUMBER を使用

重複排除には、FIRST_VALUE や LAST_VALUE の代わりに ROW_NUMBER 関数を使用してください。ROW_NUMBER を使用すると、Deduplicate オペレーターはキーごとに最初または最新のレコードのみを格納し、集約関数が必要とする完全なステートを維持する必要がありません。

多次元集約に FILTER 構文を使用

多次元集約には、CASE WHEN の代わりに FILTER 構文を使用してください。例えば、モバイルデバイス、デスクトップデバイス、およびすべてのデバイスでのユニークビジター数をカウントする場合です。SQL オプティマイザーは、同じキーに対する異なるフィルター引数を認識し、複数の COUNT DISTINCT 計算でステートデータを共有します。これにより、ステートへのアクセス回数が削減されます。テストによると、FILTER 構文は CASE WHEN 構文の 2 倍のパフォーマンスを提供できます。

複数ストリーム結合における結合順序の調整

Apache Flink は、データストリームを処理するためにバイナリハッシュ結合を使用します。複数のストリームを結合する場合、結合が追加されるたびにステートの冗長性が増幅されます。

小さいストリームを大きいストリームの前に結合することで、ステートの冗長性による増幅効果を緩和し、処理効率を向上させます。

ディスク読み取りの最小化

ディスクストレージへのアクセスを減らして、システムパフォーマンスを向上させ、メモリ使用率を最適化します。具体的な手法については、「ディスク読み取りの最小化」をご参照ください。

オプティマイザーによって派生するオペレーター

オプティマイザーは、以下のステートフルオペレーターを導入する場合があります。これら 3 つすべてが、ステートのクリーンアップに TTL (Time-to-Live) を使用します。

ChangelogNormalize

ChangelogNormalize オペレーターは、プライマリキーを含むチェンジログを処理し、効率、データ整合性、およびデータの正確性を保証します。これは 2 つのシナリオで使用されます。

シナリオ 1: ソーステーブルにプライマリキーがあり、UPSERT 操作をサポートしている。

このタイプのテーブルは、アップサートソーステーブルとして知られています。プライマリキーのシーケンスを維持しながら、プライマリキーに対する UPDATE (INSERT および UPDATE_AFTER) と DELETE 操作のみを含むチェンジログストリームを生成します。例えば、Upsert Kafka コネクタはアップサートソーステーブルを作成します。カスタムソースコネクタも、getChangelogMode メソッドをオーバーライドすることで UPSERT 操作をサポートできます。

@Override
public ChangelogMode getChangelogMode() {
    return ChangelogMode.upsert();
}

シナリオ 2: 'table.exec.source.cdc-events-duplicate' = 'true' 構成が指定されている。

Change Data Capture (CDC) の at-least-once 処理は、重複したチェンジログレコードを生成する可能性があります。exactly-once 処理を実現するには、この構成を指定して重複を削除します。次の図は、このシナリオを示しています。

この例では、入力データはソーステーブルの DDL 文で定義されたプライマリキーに基づいてハッシュシャッフルされます。ChangelogNormalize オペレーターは ValueState オブジェクトを作成し、各プライマリキーの最新レコードを格納します。次の図は、オペレーターがステートを更新し、出力を生成する方法を示しています。2 番目の -U(2, 'Jerry', 77) レコードが到着すると、ステートに格納されている対応する値は空になります。これは、プライマリキー 2 の ADD 変更 (+I および +UA) の数が RETRACT 変更 (-D および -UB) の数と等しいことを意味します。重複レコードは破棄されます。

SQL によって呼び出されるオペレーター

これらのオペレーターは、SQL 文の構文によって直接トリガーされます。TTL またはウォーターマークの進行状況に基づいてステートをクリーンアップします。

ウィンドウベースのオペレーター (WindowAggregate、WindowDeduplicate、WindowJoin、WindowTopN) は、ウォーターマークの進行状況に基づいてステートをクリーンアップします。ウォーターマークのタイムスタンプがウィンドウの終了時間を超えると、組み込みのタイマーがステートのクリーンアップをトリガーします。