Realtime Compute for Apache Flink:PostgreSQL CDC

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

Flink PostgreSQL CDC ジョブは、レプリケーションスロットを使用して、WAL の早期パージを防ぎ、データ整合性を確保します。スロットの管理が不適切な場合、過剰なディスク使用量や読み取り遅延を引き起こす可能性があります。ベストプラクティスは次のとおりです：

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

  • Flink は、WAL データの損失を防ぐため、ジョブが停止したり、ステートレスで再起動したりした場合でも、レプリケーションスロットを自動的に削除しません。

  • ジョブを再開しない場合は、手動でそのレプリケーションスロットを削除してディスク領域を解放してください。

    説明

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

• 古いスロットの再利用の回避

  • 常に新しいスロット名を使用してください。古いスロットを再利用すると、ジョブは起動時に蓄積された履歴 WAL データを読み取らなければならなくなり、新しいデータの処理が遅延します。

  • PostgreSQL では、接続ごとに 1 つのスロットが必要です。各ジョブは一意のスロット名を使用する必要があります。

    説明

    命名規則：slot.name をカスタマイズする場合、一時スロットとの競合を避けるため、my_slot_1 のような数字のサフィックスを持つ名前は避けてください。

• 増分スナップショット有効時のスロットの動作

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

  • スロット作成ルール：

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

    • 増分スナップショットが有効な場合：

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

      • 増分フェーズ：すべての一時スロットは自動的に回収されます。1 つのグローバルスロットのみが保持されます。

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

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

  • 利用可能なスロットまたはディスク領域が限られている場合は、スナップショットの並列度を下げて、使用する一時スロットの数を減らしてください。これにより、スナップショットの読み取り速度が低下します。

  • ダウンストリームのシンクがべき等な書き込みをサポートしている場合は、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	チェックポイント間の間隔。	単位は 10min や 30s のような期間形式です。
  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 という名前でパブリケーションを作成し、すべてのテーブルを含める (または、指定したテーブルを対象に、ジョブごとにパブリケーションを作成)
   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: 120s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

コネクタオプション