Realtime Compute for Apache Flink:順序が乱れたチェンジログイベントの処理

一般的なユースケース

The SinkUpsertMaterializer オペレーターは、次のシナリオで使用されます:

• シンクテーブルにはプライマリキーがありますが、テーブルに書き込まれたデータが UNIQUE 制約を満たしていません。考えられる原因には、次の操作が含まれますが、これらに限定されません。

  • ソーステーブルにプライマリキーがない場合に、シンクテーブルのプライマリキーを定義します。

  • データをシンクテーブルに挿入するときにソーステーブルのプライマリキー列を含めないか、ソーステーブルのプライマリキー以外の列をシンクテーブルのプライマリキーとして使用します。

  • ソーステーブルのプライマリキー列で使用されるデータ型が、変換またはグループ集計後に精度が低下します。たとえば、列が BIGINT 型から INT 型に変換されます。

  • ソーステーブルのプライマリキー列またはグループ集計によって生成された一意キーに対してデータ変換を実行します。たとえば、複数のプライマリキーを連結またはマージして単一の列にします。

    CREATE TABLE students (
      student_id BIGINT NOT NULL,
      student_name STRING NOT NULL,
      course_id BIGINT NOT NULL,
      score DOUBLE NOT NULL,
      PRIMARY KEY(student_id) NOT ENFORCED
    ) WITH (...);
    
    CREATE TABLE performance_report (
      student_info STRING NOT NULL PRIMARY KEY NOT ENFORCED,
      avg_score DOUBLE NOT NULL
    ) WITH (...);
    
    CREATE TEMPORARY VIEW v AS
    SELECT student_id, student_name, AVG(score) AS avg_score
    FROM students
    GROUP BY student_id, student_name;
    
    -- 連結された結果は UNIQUE 制約を満たさなくなりましたが、シンクテーブルのプライマリキーとして使用されます。
    INSERT INTO performance_report
    SELECT 
      CONCAT('id:', student_id, ',name:', student_name) AS student_info,
      avg_score
    FROM v;

• このトピックの例に示すように、入力データの元のソートがシンクテーブルに書き込まれる前に中断されます。テーブル s1 と s2 で実行される結合操作は、s1 のプライマリキーに基づいていませんが、シンクテーブルはテーブル s1 と同じプライマリキーを持っています。これにより、データのソートが中断されます。

• table.exec.sink.upsert-materialize パラメーターが 'force' に設定されています。詳細については、このトピックの「パラメーター設定」セクションをご参照ください。

使用上の注意

SinkUpsertMaterializer 演算子は、状態に RowData 値のリストを保持します。これにより、状態サイズが増加し、状態データにアクセスするための追加の I/O オーバーヘッドが必要になり、スループットに影響を与える可能性があります。SinkUpsertMaterializer 演算子の使用は避けることをお勧めします。

パラメータ設定

table.exec.sink.upsert-materialize パラメータを使用して、SinkUpsertMaterializer 演算子を設定します。有効な値：

• auto（デフォルト）：Flink は、変更ログイベントの順序が正しくないかどうかを推測し、必要に応じて SinkUpsertMaterializer 演算子を追加します。

• none：SinkUpsertMaterializer 演算子を使用しません。

• force：常に SinkUpsertMaterializer 演算子を使用します。この場合、シンクテーブルにプライマリキーが指定されていない場合でも、演算子が追加されます。これにより、データのマテリアライゼーションが保証されます。

パラメータを auto に設定した場合、SinkUpsertMaterializer 演算子の追加は、必ずしもイベントの順序が正しくないことを意味するわけではないことに注意してください。たとえば、GROUPING SETS 句と COALESCE 関数を一緒に使用して null 値を変換する場合、SQL プランナーは、生成された upsert キーがシンクテーブルのプライマリキーと一致するかどうかを判断できない場合があります。この場合、Flink は結果の正確性を確保するために SinkUpsertMaterializer 演算子を追加します。ただし、SinkUpsertMaterializer 演算子なしで最終結果が正しい場合は、table.exec.sink.upsert-materialize パラメータを none に設定することをお勧めします。

SinkUpsertMaterializer の使用を避ける

SinkUpsertMaterializer 演算子の使用を避けるには、次の項目に注意してください。

• 重複除去やグループ集計などの操作で使用されるパーティションキーが、シンクテーブルのプライマリキーと同じであることを確認します。

• 単一の並列度がデータセットに適している場合に順序が乱れたデータを防ぐには、並列度を 1 に設定します。このシナリオでは、table.exec.sink.upsert-materialize を none に設定して SinkUpsertMaterializer オペレーターを無効にします。

• シンク演算子とアップストリーム演算子（重複除去演算子やグループ集計演算子など）の演算子チェーンが作成され、6.0 より前の VVR バージョンを使用した場合にデータ精度の問題が発生しない場合は、デプロイを移行して VVR 6.0 以降を使用できます。table.exec.sink.upsert-materialize パラメータを none に設定し、他の設定を保持してください。デプロイの移行方法の詳細については、デプロイのエンジンバージョンのアップグレード をご参照ください。

SinkUpsertMaterializer 演算子を使用する必要がある場合は、次の項目に注意してください。

• データをシンクテーブルに書き込むときに、CURRENT_TIMESTAMP や NOW などの非決定性関数によって生成された列を追加しないでください。これにより、upsert キーが使用できない場合に SinkUpsertMaterializer 演算子の状態の異常な膨張を防ぎます。

• SinkUpsertMaterializer 演算子の状態が大きく、パフォーマンスに影響を与える場合は、デプロイの並列度を上げます。詳細については、デプロイのリソースの設定 をご参照ください。

既知の問題

SinkUpsertMaterializer 演算子は、順序が正しくない変更ログイベントの問題を解決しますが、次の理由により、状態サイズが継続的に増加する可能性があります。

• 状態の Time To Live（TTL）が設定されていないか、状態の TTL が長すぎるため、状態の保持期間が長すぎます。ただし、TTL が短すぎると、FLINK-29225 の問題で説明されているように、状態に不要なダーティデータが保持される可能性があります。この問題は、DELETE イベントと対応する ADD イベントの間の時間間隔が設定された TTL を超えた場合に発生します。この場合、Flink はログに次の警告メッセージを出力します。

  int index = findremoveFirst(values, row);     
  if (index == -1) {          
      LOG.info(STATE_CLEARED_WARN_MSG);          
      return;     
  }

  ビジネス要件に基づいて TTL を設定することをお勧めします。詳細については、デプロイの設定 をご参照ください。VVR 8.0.7 以降を使用する Realtime Compute for Apache Flink では、演算子ごとに異なる TTL 値を設定して、大規模状態デプロイのリソース消費を削減できます。詳細については、演算子の並列度、チェーン戦略、および TTL の設定 をご参照ください。

• SinkUpsertMaterializer 演算子によって受信された更新ストリームの upsert キーを推定できず、更新ストリームに非決定性列が含まれている場合、履歴データを予期どおりに削除できません。これにより、状態サイズが継続的に増加します。

参考資料

Realtime Compute for Apache Flink と Apache Flink のエンジンバージョンのマッピングについては、リリースノート をご参照ください。