Realtime Compute for Apache Flink：Postgres CDC（公測中）

背景資訊

特色功能

Postgres CDC連接器接入CDC增量快照框架（Realtime Compute引擎VVR 8.0.6及以上版本）。Postgres CDC讀取歷史全量資料後，自動切換到WAL變更日誌讀取，保證不多讀也不少讀資料。即使發生故障，也能保證Exactly Once語義處理資料。Postgres CDC源表提供了並發讀取全量資料，無鎖讀取和斷點續傳的能力。

作為源表，功能與優勢詳情如下：

• 流批一體，支援讀取全量和增量資料，無需維護兩套流程。

• 支援並發讀取全量資料，效能水平擴充。

• 全量讀取無縫切換增量讀取，自動縮容，節省計算資源。

• 全量階段讀取支援斷點續傳，更穩定。

• 無鎖讀取全量資料，不影響線上業務。

前提條件

Postgres CDC連接器通過PostgreSQL資料庫的邏輯複製讀取CDC變更流資料，支援阿里雲RDS PostgreSQL、Amazon RDS PostgreSQL和自建PostgreSQL。

完成配置後確保有下列的條件：

• wal_level參數的值需設定為logical，即在預寫式日誌WAL（Write-ahead logging）中增加支援邏輯編碼所需的資訊。

• 訂閱表的REPLICA IDENTITY為FULL（發出的插入和更新操作事件包含表中所有列的舊值），以保障該表資料同步的一致性。

  說明

  REPLICA IDENTITY是PostgreSQL特有的表級設定，它決定了邏輯解碼外掛程式在發生（INSERT）和更新（UPDATE）事件時，是否包含涉及的表列的舊值。REPLICA IDENTITY取值含義詳情請參見REPLICA IDENTITY。

• 需要確保max_wal_senders和max_replication_slots的參數值均大於當前資料庫複寫槽已使用數與Flink作業所需要的slot數量。

• 確保賬戶系統許可權為SUPERUSER或者同時擁有LOGIN和REPLICATION許可權，並且具有訂閱表的SELECT許可權用於全量資料查詢。

注意事項

• 僅Realtime Compute引擎8.0.6及以上版本支援Postgres CDC增量快照功能。

• 請及時管理Replication Slot，以免出現磁碟空間浪費的問題。

  為了防止在Flink作業重啟過程中由於Checkpoint對應的WAL（Write-Ahead Log）段被清除而引發資料丟失，Flink作業不會自動移除Replication Slot。因此，如果確認特定的Flink作業不會再次啟動，應當手動刪除相關的Replication Slot，以釋放其佔用的資源。另外，如果PostgreSQL的Replication Slot的確認位點長時間不向前推進，PostgreSQL不會清理該槽位點之後的WAL條目，這可能會導致未使用的WAL積累而佔用過多的磁碟空間。

• 開啟增量快照時，Postgres CDC連接器必須開啟Checkpoint，並且Source表必須聲明主鍵。Source多並發讀取全量資料時會建立多個臨時的Replication Slot。

  不開啟增量快照讀取的PostgreSQL CDC Source僅支援單一併發，因此只需要一個全域Slot。當開啟增量快照時，PostgreSQL CDC Source在全量階段所需的最大Slot數量為Source數量 * 並發數 + 1。進入增量階段後，系統自動回收在全量階段建立的Slot，僅保留一個全域Slot。如果Slot數量有限，需要控制全量階段的並發數量，這樣做的缺點是會降低讀取速度。如果下遊運算元或儲存支援等冪性，可以啟用scan.incremental.snapshot.backfill.skip = true以跳過全量階段的日誌讀取，這樣做的缺點是僅能提供至少一次（At-Least Once）的語義保證。如果SQL要做彙總、關聯等操作，不建議跳過全量階段日誌的讀取。

• 不開啟增量快照時，Postgres CDC連接器不支援在全表掃描階段執行Checkpoint。

  不開啟增量快照時，如果您的作業在全表掃描階段觸發Checkpoint，則可能由於Checkpoint逾時導致作業Failover。因此，建議您在其他配置中配置如下參數，具體操作請參見空間管理與操作。避免在全量同步階段由於Checkpoint逾時導致Failover。

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

  相關的參數說明詳情如下表所示。

  參數	說明	備忘
  execution.checkpointing.interval	Checkpoint的時間間隔。	單位是Duration類型，例如10min或30s。
  execution.checkpointing.tolerable-failed-checkpoints	容忍Checkpoint失敗的次數。	該參數的取值與Checkpoint調度間隔時間的乘積就是允許的快照讀取時間。 說明 如果表特別大，建議將該參數值配置得大一些。
  restart-strategy	重啟策略。	參數取值如下： fixed-delay：固定延遲重啟策略。 failure-rate：故障率重啟策略。 exponential-delay：指數延遲重啟策略。 詳情請參見Restart Strategies。
  restart-strategy.fixed-delay.attempts	固定延遲重啟策略下，嘗試重啟的最大次數。	無。

文法結構

CREATE TABLE postgrescdc_source (
  shipment_id INT,
  order_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<yourHostname>',
  'port' = '5432',
  'username' = '<yourUserName>',
  'password' = '<yourPassWord>',
  'database-name' = '<yourDatabaseName>',
  'schema-name' = '<yourSchemaName>',
  'table-name' = '<yourTableName>'
);

WITH參數

類型映射

使用樣本

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;

相關文檔

• Realtime ComputeFlink版支援的連接器列表，請參見支援的連接器。

• 將資料寫入PolarDB PostgreSQL版（Oracle文法相容1.0）結果表，請參見PolarDB PostgreSQL版（Oracle文法相容1.0）。

• 如果您需要讀寫RDS MySQL、PolarDB for MySQL或者自建MySQL資料庫，請使用MySQL連接器。