Realtime Compute for Apache Flink:PostgreSQL CDC

レプリケーションスロット

Flink PostgreSQL CDC ジョブは、レプリケーションスロットを使用して、先行書き込みログ (WAL) が早期にパージされるのを防ぎ、データ整合性を確保します。レプリケーションスロットが不適切に管理されていると、過剰なディスク使用量やデータ読み取りの遅延などの問題を引き起こす可能性があります。以下のベストプラクティスに従ってください：

• 未使用スロットの速やかなクリーンアップ

  • Flink は、ジョブが停止したり、ステートレスな再起動を実行したりした後でも、レプリケーションスロットを自動的に削除しません。これは、WAL がパージされた場合に発生する可能性のあるデータ損失を防ぐためです。

  • ジョブが再起動されないことを確認した場合は、関連するレプリケーションスロットを手動で削除して、ディスク領域を解放する必要があります。

    説明

    ライフサイクル管理： レプリケーションスロットをジョブレベルのリソースとして扱い、ジョブの開始と停止に合わせて管理してください。

• 古いスロットの再利用を避ける

  • 新しいジョブでは、古いスロット名を再利用するのではなく、新しいスロット名を使用する必要があります。古いスロットを再利用すると、ジョブが起動時に大量の履歴 WAL データを読み取ることになり、新しいデータの処理が遅延する可能性があります。

  • PostgreSQL の論理レプリケーションでは、1 つのスロットは 1 つの接続でしか使用できません。異なるジョブは、異なるスロット名を使用する必要があります。

    説明

    命名規則： slot.name をカスタマイズする際、一時的なスロットとの競合を避けるため、my_slot_1 のように末尾に数字が付く名前は避けてください。

• インクリメンタルスナップショットが有効な場合のスロットの動作

  • 前提条件： チェックポイントが有効であり、ソーステーブルに主キーが定義されている必要があります。

  • スロット作成ルール：

    • インクリメンタルスナップショットが無効な場合： 並列度 1 のみサポートされます。1 つのグローバルスロットが使用されます。

    • インクリメンタルスナップショットが有効な場合：

      • スナップショットフェーズ： 各並行ソースサブタスクは一時的なスロットを作成します。命名形式は ${slot.name}_${task_id} です。

      • インクリメンタルフェーズ： すべての一時的なスロットは自動的に回収されます。グローバルスロットが 1 つだけ保持されます。

  • 最大スロット数： ソースの並列度 + 1 (スナップショットフェーズ中)

• リソースとパフォーマンス

  • PostgreSQL で利用可能なスロット数やディスク容量に制限がある場合は、スナップショットフェーズの並列度を下げて、使用する一時スロットの数を減らしてください。これにより、スナップショットフェーズ中の読み取り速度は低下します。

  • ダウンストリームシンクがべき等な書き込みをサポートしている場合は、scan.incremental.snapshot.backfill.skip = true を設定してください。この設定により、スナップショットフェーズ中の WAL バックフィルがスキップされ、ジョブの起動が高速化されます。

    この設定は、at-least-once セマンティクスのみを提供します。中間状態に必要な履歴の変更が失われる可能性があるため、集約やルックアップ結合などのステートフルな計算には適していません。

• インクリメンタルスナップショットが無効な場合、スナップショットフェーズではチェックポイントはサポートされません。

  スナップショットフェーズ中のタイムアウトを回避するための設定

  インクリメンタルスナップショットが無効な場合、ジョブがスナップショットフェーズ中にチェックポイントをトリガーすると、チェックポイントタイムアウトによりジョブがフェイルオーバーする可能性があります。これを防ぐには、[Other Configuration] で次のパラメータを設定してください。詳細については、「カスタム実行パラメータの設定」をご参照ください。この設定により、スナップショットフェーズ中のチェックポイントタイムアウトによるフェイルオーバーを防ぐことができます。

  execution.checkpointing.interval: 10min
  execution.checkpointing.tolerable-failed-checkpoints: 100
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 2147483647

  次の表で各パラメータを説明します。

  パラメータ	説明	備考
  execution.checkpointing.interval	チェックポイント間の間隔。	単位は 10 min や 30 s などの期間値です。
  execution.checkpointing.tolerable-failed-checkpoints	ジョブが失敗するまでに許容されるチェックポイントの失敗回数。	このパラメータとチェックポイント間隔の積が、許容されるスナップショット読み取り時間になります。 説明 テーブルが非常に大きい場合は、このパラメータをより大きな値に設定してください。
  restart-strategy	ジョブの再起動戦略。	有効な値： fixed-delay： 固定遅延再起動戦略。 failure-rate： 障害率再起動戦略。 exponential-delay： 指数関数的遅延再起動戦略。 詳細については、「再起動戦略」をご参照ください。
  restart-strategy.fixed-delay.attempts	fixed-delay 再起動戦略の最大再起動試行回数。	–

PostgreSQL パブリケーションの再利用

PostgreSQL CDC コネクタは、パブリケーションに依存して、どのテーブルの変更をスロットにプッシュするかを決定します。複数のジョブが同じパブリケーションを共有すると、それらの設定は上書きされます。

原因

publication.autocreate.mode のデフォルト値は filtered で、これにはコネクタ設定で指定されたテーブルのみが含まれます。このモードでは、ジョブの開始時にパブリケーション内のテーブルが変更されるため、他のジョブの読み取り操作に影響を与える可能性があります。

解決策

1. 監視対象のすべてのテーブルを含むパブリケーションを PostgreSQL で手動で作成してください。または、ジョブごとに個別のパブリケーションを作成してください。

   -- my_flink_pub という名前で、テーブル table_a および table_b のパブリケーションを作成します
   CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
   -- または、データベース内のすべてのテーブルのパブリケーションを作成します
   CREATE PUBLICATION my_flink_pub FOR ALL TABLES;

   説明

   大規模なデータベースの場合、すべてのテーブルをサブスクライブすることは推奨されません。これは、過剰なネットワーク帯域幅の使用や Flink クラスターでの高いCPU使用率を引き起こす可能性があるためです。

2. 次の Flink 設定を追加してください：

   • debezium.publication.name = 'my_flink_pub' (パブリケーション名を指定します)

   • debezium.publication.autocreate.mode = 'disabled' (起動時に Flink がパブリケーションを作成または変更しようとするのを防ぎます)

このアプローチは完全なアイソレーションを提供し、パブリケーションを手動で管理できるようにします。これにより、新しいジョブが既存のジョブに影響を与えるのを防ぎ、より安全なアクセス制御が可能になります。

構文

CREATE TABLE postgrescdc_source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- バックフィルをスキップすると読み取りが高速化され、リソース使用量を削減できますが、データが重複する可能性があります。ダウンストリームのシンクがべき等である場合に有効にしてください。
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- 本番環境では、これを 'filtered' または 'disabled' に設定し、Flink 経由ではなく手動でパブリケーションを管理してください。
  'debezium-publication.autocreate.mode' = 'disabled'
  -- 複数のソースがある場合は、ソースごとに異なるパブリケーションを設定してください。
  --'debezium.publication.name' = 'my_flink_pub'
);

コネクタオプション

型マッピング

例

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

Flink CDC

VVR V11.4 以降では、PostgreSQL コネクタを Flink CDC ソースとしてサポートしています。

構文

source:
  type: postgres
  name: PostgreSQL Source
  hostname: localhost
  port: 5432
  username: pg_username
  password: pg_password
  tables: db.scm.tbl
  slot.name: test_slot
  scan.startup.mode: initial
  server-time-zone: UTC
  connect.timeout: 30s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

コネクタオプション