PolarDB-X - Realtime Compute for Apache Flink

本文為您介紹如何使用PolarDB-X連接器。

背景資訊

PolarDB 分布式版（PolarDB for Xscale，簡稱“PolarDB-X”）是阿里雲自主設計研發的高效能雲原生分散式資料庫產品，為使用者提供高吞吐、大儲存、低延時、易擴充和超高可用的雲時代資料庫服務。

重要

僅支援VVR 11.5及以上版本，配合PolarDB-X 2.0及更高版本使用。

目前，PolarDB-X CDC連接器只支援作為源表使用。如您需要對PolarDB-X執行個體進行維表查詢、或作為結果表寫入，請使用MySQL連接器（公測中）。

類別	詳情
支援類型	源表，資料攝入資料來源
運行模式	僅支援流模式
資料格式	暫不適用
特有監控指標	currentFetchEventTimeLag：資料產生到拉取到Source Operator的間隔。該指標僅在Binlog階段有效，Snapshot階段該值恒為0。 currentEmitEventTimeLag：資料產生到離開Source Operator的間隔。該指標僅在Binlog階段有效，Snapshot階段該值恒為0。 sourceIdleTime：源表至今有多久未產生新資料。
API種類	SQL，資料攝入YAML作業
是否支援更新或刪除結果表資料	否

特色功能

PolarDB-X CDC連接器針對Binlog解析階段進行效能最佳化，支援PolarDB-X伺服器側對無關Binlog進行過濾和裁剪，從而提升輸送量、節省網路頻寬。

Binlog按需訂閱樣本

此版本支援在服務端對Binlog進行過濾、只發送所需的變更日誌給用戶端，從而起到降低網路流量壓力、提升日誌消費吞吐的最佳化作用。

例如，如果您只需訂閱PolarDB-X伺服器中db.table1和db.table2表的變更資料，可以像這樣配置Flink SQL作業：

CREATE TABLE polardbx_table_foo (
  ... -- 在這裡定義表結構
) WITH (
  'connector' = 'polardbx-cdc',
  'database-name' = 'db',
  'table-name' = '.*',
  ..., -- 其他參數
  'polardbx.binlog.include.tables' = 'db.table1,db.table2' -- 只訂閱對應表的資料
);

相較於MySQL CDC連接器會將整個執行個體中全量的變更Binlog日誌載入到本地、並在用戶端進行過濾，PolarDB-X CDC連接器具備服務端過濾Binlog、用戶端按需訂閱Binlog的能力，能夠大幅減少網路IO開銷。

使用限制

如您需要使用服務端Binlog服務端過濾、按表訂閱功能，需要保證PolarDB-X服務端版本為2.5.0及以上，且Log Service組件為5.4.20或更高版本。

SQL

文法結構

CREATE TABLE polardbx_customer_table(
  `id` STRING,
  [columnName dataType,]*
  PRIMARY KEY(`id`) NOT ENFORCED
) WITH (
  'connector' = 'polardbx-cdc',
  'hosts' = 'pxc-**************-pub.polarx.rds.aliyuncs.com',
  'username' = 'pdx_user',
  'password' = 'pdx_password',
  'database' = 'full_db',
  'collection' = 'customers'
)

WITH參數

參數	說明	資料類型	是否必填	預設值	備忘
connector	連接器名稱。	STRING	是	無	固定值為polardbx-cdc。
hostname	PolarDB-X資料庫的IP地址或者Hostname。	STRING	是	無	建議填寫執行個體串連資訊中的叢集地址。
port	PolarDB-X資料庫的服務連接埠號碼。	INTEGER	否	3306	無。
username	PolarDB-X資料庫服務的使用者名稱。	STRING	是	無	無。
password	PolarDB-X資料庫服務的密碼。	STRING	是	無	無。
database-name	PolarDB-X資料庫名稱。	STRING	是	無	支援使用Regex匹配讀取多個資料庫的資料。說明使用Regex時，不要使用^和$符號匹配開頭和結尾。
table-name	PolarDB-X表名。	STRING	是	無	支援使用Regex匹配讀取多張表的資料。說明使用Regex時，不要使用^和$符號匹配開頭和結尾。
server-time-zone	資料庫在使用的會話時區。	STRING	否	作業運行環境可用性區域時區。	指定 IANA 時區標識符，例如 Asia/Shanghai。該參數控制了源表中的TIMESTAMP類型如何轉換為STRING類型。
scan.incremental.snapshot.chunk.size	增量快照分塊讀取時，每個chunk的大小（包含的行數）。	INTEGER	否	8096	PolarDB-X 將表切分為多個分區（Chunk）進行讀取，並在記憶體中緩衝分區資料。減少單分區行數會增加分區總量。這雖然細化了故障恢複粒度，但也增加了記憶體溢出（OOM）風險並降低輸送量。請配置合理的分區大小以平衡效能。
scan.snapshot.fetch.size	當讀取表的全量資料時，每次最多拉取的記錄數。	INTEGER	否	1024	無。
connect.timeout	串連PolarDB-X資料庫伺服器逾時時，重試串連之前等待逾時的最長時間。	DURATION	否	30s	無。
connection.pool.size	資料庫連接池大小。	INTEGER	否	20	資料庫連接池用於複用串連，可以降低資料庫連接數量。
connect.max-retries	串連PolarDB-X資料庫時，串連失敗後重試的最大次數。	INTEGER	否	3	無。
scan.startup.mode	消費資料時的啟動模式。	STRING	否	initial	參數取值如下： initial（預設）：在第一次啟動時，會先掃描歷史全量資料，然後讀取最新的Binlog資料。 latest-offset：在第一次啟動時，不會掃描歷史全量資料，直接從Binlog的末尾（最新的Binlog處）開始讀取，即唯讀取該連接器啟動以後的最新變更。 earliest-offset：不掃描歷史全量資料，直接從可讀取的最早Binlog開始讀取。 specific-offset：不掃描歷史全量資料，從您指定的Binlog位點啟動，位點可通過同時配置scan.startup.specific-offset.file和scan.startup.specific-offset.pos參數來指定從特定Binlog檔案名稱和位移量啟動，也可以只配置scan.startup.specific-offset.gtid-set來指定從某個GTID集合啟動。 timestamp：不掃描歷史全量資料，從指定的時間戳記開始讀取Binlog。時間戳記通過scan.startup.timestamp-millis指定，單位為毫秒。重要對於earliest-offset，specific-offset和timestamp啟動模式，啟動時的表結構必須與指錨點一致。結構不匹配將導致作業報錯。請確保在指定 Binlog 位點至作業啟動期間，表結構未發生變更。
scan.startup.specific-offset.file	使用指錨點模式啟動時，啟動位點的Binlog檔案名稱。	STRING	否	無	使用該配置時，scan.startup.mode必須配置為specific-offset。檔案名稱格式例如`mysql-bin.000003`。
scan.startup.specific-offset.pos	使用指錨點模式啟動時，啟動位點在指定Binlog檔案中的位移量。	INTEGER	否	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
scan.startup.specific-offset.gtid-set	使用指錨點模式啟動時，啟動位點的GTID集合。	STRING	否	無	使用該配置時，scan.startup.mode必須配置為specific-offset。GTID集合格式例如`24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`。
scan.startup.timestamp-millis	使用指定時間模式啟動時，啟動位點的毫秒時間戳記。	LONG	否	無	使用該配置時，scan.startup.mode必須配置為timestamp。時間戳記單位為毫秒。
scan.startup.specific-offset.skip-events	從指定的位點讀取時，跳過多少Binlog事件。	INTEGER	否	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
scan.startup.specific-offset.skip-rows	從指定的位點讀取時，跳過多少行變更（一個Binlog事件可能對應多行變更）。	INTEGER	否	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
heartbeat.interval	Source通過心跳事件推動Binlog位點前進的時間間隔。	DURATION	否	無	心跳事件強制推進 Source 端的 Binlog 位點。此機制防止低頻更新導致 Binlog 到期。Binlog 到期將引發作業失敗，且僅能通過無狀態重啟恢複。
chunk-meta.group.size	chunk元資訊的大小。	INTEGER	否	1000	如果元資訊大於該值，元資訊會分為多份傳遞。
chunk-key.even-distribution.factor.upper-bound	是否可以均勻分區的chunk分布因子的上限。	DOUBLE	否	1000.0	分布因子大於該值會使用非均勻分區。 chunk分布因子 = (MAX(chunk-key) - MIN(chunk-key) + 1) / 總資料行數。
chunk-key.even-distribution.factor.lower-bound	是否可以均勻分區的chunk分布因子的下限。	DOUBLE	否	0.05	分布因子小於該值會使用非均勻分區。 chunk分布因子 = (MAX(chunk-key) - MIN(chunk-key) + 1) / 總資料行數。
scan.newly-added-table.enabled	從Checkpoint重啟時，是否掃描新增的捕獲表。	BOOLEAN	否	false	啟用後，系統將同步此前未匹配的新增表，並從狀態中移除不再匹配的表。從Checkpoint或Savepoint重啟時生效。
scan.incremental.snapshot.chunk.key-column	指定快照階段用於資料分區的列。	STRING	見備忘	無	無主鍵表必填，選擇的列必須是非空類型（NOT NULL）。有主鍵的表為選填，僅支援從主鍵中選擇一列。
scan.incremental.close-idle-reader.enabled	是否在快照結束後關閉閒置 Reader。	BOOLEAN	否	false	該配置生效需要同時設定execution.checkpointing.checkpoints-after-tasks-finish.enabled為true。
scan.incremental.snapshot.backfill.skip	是否在快照讀取階段跳過backfill。	BOOLEAN	否	false	參數取值如下： true：快照讀取階段跳過backfill。 false（預設）：快照讀取階段不跳過backfill。如果跳過backfill，快照階段表的更改將在稍後的增量階段讀取，而不是合并到快照中。重要跳過backfill可能導致資料不一致，因為快照階段發生的變更可能會被重放，僅保證at-least-once語義。
scan.parse.online.schema.changes.enabled	在增量階段，是否嘗試解析 RDS 無鎖變更 DDL 事件。	BOOLEAN	否	false	參數取值如下： true：解析 RDS 無鎖變更 DDL 事件。 false（預設）：不解析 RDS 無鎖變更 DDL 事件。實驗性功能。建議在執行線上無鎖變更前，先對Flink作業建立一個Savepoint以便恢複。
scan.only.deserialize.captured.tables.changelog.enabled	在增量階段，是否僅對指定表的變更事件進行還原序列化。	BOOLEAN	否	true	參數取值如下： true：僅對目標表的變更資料進行還原序列化，加快Binlog讀取速度。 false（預設）：對所有表的變更資料進行還原序列化。
scan.read-changelog-as-append-only.enabled	是否將changelog資料流轉換為append-only資料流。	BOOLEAN	否	false	參數取值如下： true：所有類型的訊息（包括INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER）都會轉換成INSERT類型的訊息。僅在需要儲存上遊表刪除訊息等特殊情境下開啟使用。 false（預設）：所有類型的訊息都保持原樣下發。
scan.parallel-deserialize-changelog.enabled	在增量階段,是否使用多線程對變更事件進行解析。	BOOLEAN	否	false	參數取值如下： true：在變更事件的還原序列化階段採用多執行緒，同時保證Binlog事件順序不變，從而加快讀取速度。 false（預設）：在事件的還原序列化階段使用單線程處理。
scan.parallel-deserialize-changelog.handler.size	多線程對變更事件進行解析時，事件處理器的數量。	INTEGER	否	2	無。
scan.incremental.snapshot.unbounded-chunk-first.enabled	快照讀取階段是否先分發無界的分區。	BOOLEAN	否	false	參數取值如下： true：快照讀取階段優先分發無界的分區。 false（預設）：快照讀取階段不優先分發無界的分區。實驗性功能。開啟後能夠降低TaskManager在快照階段同步最後一個分區時遇到記憶體溢出 (OOM) 的風險，建議在作業第一次啟動前添加。
polardbx.binlog.ignore.archive-events.enabled	是否忽略 PolarDB-X Binlog 中的歸檔事件（主要是 `DELETE` 事件）。	BOOLEAN	否	false
polardbx.binlog.ignore.query-events.enabled	是否忽略 PolarDB-X Binlog 中的ROWS_QUERY_LOG_EVENT事件。	BOOLEAN	否	false
polardbx.binlog.include.tables	僅訂閱這些表的Binlog日誌。多個表名之間用逗號（`,`）分隔。	STRING	否	無
polardbx.binlog.exclude.tables	不訂閱這些表的Binlog日誌。多個表名之間用逗號（`,`）分隔。	STRING	否	無

類型映射

PolarDB-X欄位類型	Flink欄位類型
TINYINT	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
INT	INT
MEDIUMINT
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT UNSIGNED ZEROFILL
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
BIGINT UNSIGNED	DECIMAL(20, 0)
BIGINT UNSIGNED ZEROFILL
SERIAL
FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
REAL [UNSIGNED] [ZEROFILL]
NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
BOOLEAN	BOOLEAN
TINYINT(1)	BOOLEAN
DATE	DATE
TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
TIMESTAMP [(p)]	TIMESTAMP [(p)]
TIMESTAMP [(p)]	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
CHAR(n)	STRING
VARCHAR(n)
TEXT
BINARY	BYTES
VARBINARY
BLOB

資料攝入

自Realtime Compute引擎11.6版本開始，PolarDB-X連接器作為資料來源可以在資料攝入YAML作業中使用。

文法結構

source:
   type: polardbx
   name: PolarDB-X Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: pdb.order_table
   # 忽略 binlog 中的歸檔事件
   polardbx.binlog.ignore.archive-events.enabled: true
   # 忽略 binlog 中的查詢事件
   polardbx.binlog.ignore.query-events.enabled: true
   # 只訂閱 pdb.order_table 的 binlog 以減小頻寬壓力
   polardbx.binlog.include.tables: pdb.order_table

sink:
  type: values

配置項

參數	說明	是否必填	資料類型	預設值	備忘
type	資料來源類型。	是	STRING	無	固定值為polardbx。
name	資料來源名稱。	否	STRING	無	無。
hostname	PolarDB-X資料庫的IP地址或者Hostname。	是	STRING	無	建議填寫Virtual Private Cloud地址。說明如果PolarDB-X與即時Flink版不在同一VPC，需要先打通跨VPC的網路或者使用公網的形式訪問，詳情請參見空間管理與操作和Flink全託管叢集如何訪問公網？。
username	PolarDB-X資料庫服務的使用者名稱。	是	STRING	無	無。
password	PolarDB-X資料庫服務的密碼。	是	STRING	無	無。
tables	需要同步的PolarDB-X資料表。	是	STRING	無	表名支援Regex以讀取多個表的資料。可以用逗號分隔多個Regex。說明 Regex中請不要使用首尾匹配字元`^`和`$`。點號用於分割資料庫名和表名，如果需要用點號匹配任一字元，需要對點號使用反斜線進行轉義。如：db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\.。
tables.exclude	需要在同步的表中排除的表。	否	STRING	無	表名支援Regex以排除多個表的資料。可以用逗號分隔多個Regex。說明點號用於分割資料庫名和表名，如果需要用點號匹配任一字元，需要對點號使用反斜線進行轉譯。如：db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\.。
port	PolarDB-X資料庫服務的連接埠號碼。	否	INTEGER	3306	無。
schema-change.enabled	是否發送Schame變更事件。	否	BOOLEAN	true	無。
jdbc.properties.*	JDBC URL中的自訂串連參數。	否	STRING	無	您可以傳遞自訂的串連參數，例如不使用SSL協議，則可配置為'jdbc.properties.useSSL' = 'false'。
debezium.*	Debezium讀取Binlog的自訂參數。	否	STRING	無	您可以傳遞自訂的Debezium參數，例如使用'debezium.event.deserialization.failure.handling.mode'='ignore'來指定解析錯誤時的處理邏輯。
scan.incremental.snapshot.chunk.size	每個chunk的大小（包含的行數）。	否	INTEGER	8096	PolarDB-X表會被切分成多個chunk讀取。在讀完chunk的資料之前，chunk的資料會先緩衝在記憶體中。每個chunk包含的行數越少，則表中的chunk的總數量越大，儘管這會降低故障恢複的粒度，但可能導致記憶體OOM和整體的輸送量降低。因此，您需要進行權衡，並設定合理的chunk大小。
scan.snapshot.fetch.size	當讀取表的全量資料時，每次最多拉取的記錄數。	否	INTEGER	1024	無。
scan.startup.mode	消費資料時的啟動模式。	否	STRING	initial	參數取值如下： initial（預設）：在第一次啟動時，會先掃描歷史全量資料，然後讀取最新的Binlog資料。 latest-offset：在第一次啟動時，不會掃描歷史全量資料，直接從Binlog的末尾（最新的Binlog處）開始讀取，即唯讀取該連接器啟動以後的最新變更。 earliest-offset：不掃描歷史全量資料，直接從可讀取的最早Binlog開始讀取。 specific-offset：不掃描歷史全量資料，從您指定的Binlog位點啟動，位點可通過同時配置scan.startup.specific-offset.file和scan.startup.specific-offset.pos參數來指定從特定Binlog檔案名稱和位移量啟動，也可以只配置scan.startup.specific-offset.gtid-set來指定從某個GTID集合啟動。 timestamp：不掃描歷史全量資料，從指定的時間戳記開始讀取Binlog。時間戳記通過scan.startup.timestamp-millis指定，單位為毫秒。重要對於earliest-offset，specific-offset和timestamp啟動模式，如果啟動時刻和指定的啟動位點時刻的表結構不同，作業會因為表結構不同而報錯。換一句話說，使用這三種啟動模式，需要保證在指定的Binlog消費位置到作業啟動的時間之間，對應表不能發生表結構變更。
scan.startup.specific-offset.file	使用指錨點模式啟動時，啟動位點的Binlog檔案名稱。	否	STRING	無	使用該配置時，scan.startup.mode必須配置為specific-offset。檔案名稱格式例如`mysql-bin.000003`。
scan.startup.specific-offset.pos	使用指錨點模式啟動時，啟動位點在指定Binlog檔案中的位移量。	否	INTEGER	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
scan.startup.specific-offset.gtid-set	使用指錨點模式啟動時，啟動位點的GTID集合。	否	STRING	無	使用該配置時，scan.startup.mode必須配置為specific-offset。GTID集合格式例如`24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`。
scan.startup.timestamp-millis	使用指定時間模式啟動時，啟動位點的毫秒時間戳記。	否	LONG	無	使用該配置時，scan.startup.mode必須配置為timestamp。時間戳記單位為毫秒。重要在使用指定時間時，PolarDB-X CDC會嘗試讀取每個Binlog檔案的初始事件以確定其時間戳記，最終定位至指定時間對應的Binlog檔案。請保證指定的時間戳記對應的Binlog檔案在資料庫中沒有被清理且可以被讀取到。
server-time-zone	資料庫在使用的會話時區。	否	STRING	如果您沒有指定該參數，則系統預設使用Flink作業運行時的環境時區作為資料庫伺服器時區，即您選擇的可用性區域所在的時區。	例如Asia/Shanghai，該參數控制了PolarDB-X中的TIMESTAMP類型如何轉成STRING類型。更多資訊請參見Debezium時間類型。
scan.startup.specific-offset.skip-events	從指定的位點讀取時，跳過多少Binlog事件。	否	INTEGER	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
scan.startup.specific-offset.skip-rows	從指定的位點讀取時，跳過多少行變更（一個Binlog事件可能對應多行變更）。	否	INTEGER	無	使用該配置時，scan.startup.mode必須配置為specific-offset。
connect.timeout	串連PolarDB-X資料庫伺服器逾時時，重試串連之前等待逾時的最長時間。	否	DURATION	30s	無。
connect.max-retries	串連PolarDB-X資料庫服務時，串連失敗後重試的最大次數。	否	INTEGER	3	無。
connection.pool.size	資料庫連接池大小。	否	INTEGER	20	資料庫連接池用於複用串連，可以降低資料庫連接數量。
heartbeat.interval	Source通過心跳事件推動Binlog位點前進的時間間隔。	否	DURATION	30s	心跳事件用於推動Source中的Binlog位點前進，這對PolarDB-X中更新緩慢的表非常有用。對於更新緩慢的表，Binlog位點無法自動前進，通過夠心跳事件可以推到Binlog位點前進，可以避免Binlog位點不前進引起Binlog位點到期問題，Binlog位點到期會導致作業失敗無法恢複，只能無狀態重啟。
scan.incremental.snapshot.chunk.key-column	可以指定某一列作為快照階段切分分區的切分列。	否。	STRING	無	僅支援從主鍵中選擇一列。
chunk-meta.group.size	chunk元資訊的大小。	否	INTEGER	1000	如果元資訊大於該值，元資訊會分為多份傳遞。
chunk-key.even-distribution.factor.lower-bound	是否可以均勻分區的chunk分布因子的下限。	否	DOUBLE	0.05	分布因子小於該值會使用非均勻分區。 chunk分布因子 = (MAX(chunk-key) - MIN(chunk-key) + 1) / 總資料行數。
chunk-key.even-distribution.factor.upper-bound	是否可以均勻分區的chunk分布因子的上限。	否	DOUBLE	1000.0	分布因子大於該值會使用非均勻分區。 chunk分布因子 = (MAX(chunk-key) - MIN(chunk-key) + 1) / 總資料行數。
scan.incremental.close-idle-reader.enabled	是否在快照結束後關閉閒置Reader。	否	BOOLEAN	false	該配置生效，需要設定`execution.checkpointing.checkpoints-after-tasks-finish.enabled`為true。
scan.only.deserialize.captured.tables.changelog.enabled	在增量階段，是否僅對指定表的變更事件進行還原序列化。	否	BOOLEAN	true	參數取值如下： true：僅對目標表的變更資料進行還原序列化，加快Binlog讀取速度。 false（預設）：對所有表的變更資料進行還原序列化。
scan.parallel-deserialize-changelog.enabled	在增量階段,是否使用多線程對變更事件進行解析。	否	BOOLEAN	false	參數取值如下： true：在變更事件的還原序列化階段採用多執行緒，同時保證Binlog事件順序不變，從而加快讀取速度。 false（預設）：在事件的還原序列化階段使用單線程處理。
scan.parallel-deserialize-changelog.handler.size	多線程對變更事件進行解析時，事件處理器的數量。	否	INTEGER	2	無。
metadata-column.include-list	需要傳給下遊的中繼資料列。	否	STRING	無	可用的中繼資料套件括`op_ts`、`es_ts`、`query_log`、`file`和`pos`，您可以使用英文逗號分隔多個中繼資料列。說明 PolarDB-X CDC YAML連接器無需也不支援添加庫名表名和`op_type`中繼資料列。您可以直接在Transform運算式中使用`__data_event_type__`來擷取變化資料類型，或在Transform運算式中使用`__schema_name__`和`__table_name__`來擷取資料庫名和表名。重要 `file`中繼資料列代表該資料所在的binlog檔案，全量階段為"", 增量階段為binlog檔案名稱；`pos`中繼資料列代表資料所在的binlog檔案中的位移量，全量階段為"0", 增量階段為資料在binlog檔案中的位移量。
scan.newly-added-table.enabled	從Checkpoint重啟時，是否同步上一次啟動時未匹配到的新增表或者移除狀態中儲存的當前不匹配的表。	否	BOOLEAN	false	從Checkpoint或Savepoint重啟時生效。
scan.binlog.newly-added-table.enabled	在增量階段，是否發送匹配到的新增表的資料。	否	BOOLEAN	false	不能與`scan.newly-added-table.enabled`同時開啟。
scan.incremental.snapshot.chunk.key-column	為某些表指定一列作為快照階段切分分區的切分列。	否	STRING	無	通過英文冒號`:`串連表名和欄位名，表示一個指定規則，表名可以使用Regex。支援定義多個指定規則，不同指定規則通過英文分號`;`分割。例如：`db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`。對於無主鍵表必填，選擇的列必須是非空類型（NOT NULL）。有主鍵的表為選填，僅支援從主鍵中選擇一列。
scan.parse.online.schema.changes.enabled	在增量階段，是否嘗試解析 RDS 無鎖變更 DDL 事件。	否	BOOLEAN	false	參數取值如下： true：解析 RDS 無鎖變更 DDL 事件。 false（預設）：不解析 RDS 無鎖變更 DDL 事件。實驗性功能。建議在執行線上無鎖變更前，先對Flink作業執行一次快照以便恢複。
scan.incremental.snapshot.backfill.skip	是否在快照讀取階段跳過backfill。	否	BOOLEAN	false	參數取值如下： true：快照讀取階段跳過backfill。 false（預設）：快照讀取階段不跳過backfill。如果跳過backfill，快照階段表的更改將在稍後的增量階段讀取，而不是合并到快照中。重要跳過backfill可能導致資料不一致，因為快照階段發生的變更可能會被重放，僅保證at-least-once語義。
treat-tinyint1-as-boolean.enabled	是否將TINYINT(1)類型當做Boolean類型處理。	否	BOOLEAN	true	參數取值如下： true（預設）：將TINYINT(1)類型當作Boolean類型處理。 false：不將TINYINT(1)類型當作Boolean類型處理。
treat-timestamp-as-datetime-enabled	是否將TIMESTAMP類型當作DATETIME類型處理。	否	BOOLEAN	false	參數取值如下： true：將PolarDB-X TIMESTAMP類型當作DATETIME類型處理，映射到CDC TIMESTAMP類型。 false（預設）：將PolarDB-X TIMESTAMP類型映射到CDC TIMESTAMP_LTZ類型。 PolarDB-X TIMESTAMP類型儲存的是UTC時間，受時區影響，PolarDB-X DATETIME類型儲存的是字面時間，不受時區影響。開啟後會根據server-time-zone將PolarDB-X TIMESTAMP類型資料轉換成DATETIME類型。
include-comments.enabled	是否同步表注釋和欄位注釋。	否	BOOELEAN	false	參數取值如下： true：同步表注釋和欄位注釋。 false（預設）：不同步表注釋和欄位注釋。開啟後會增加作業記憶體使用量量。
scan.incremental.snapshot.unbounded-chunk-first.enabled	快照讀取階段是否先分發無界的分區。	否	BOOELEAN	false	參數取值如下： true：快照讀取階段優先分發無界的分區。 false（預設）：快照讀取階段不優先分發無界的分區。實驗性功能。開啟後能夠降低TaskManager在快照階段同步最後一個分區時遇到記憶體溢出 (OOM) 的風險，建議在作業第一次啟動前添加。
binlog.session.network.timeout	Binlog串連的網路逾時時間。	否	DURATION	10m	設值為0s時，將會使用服務端的預設逾時時間。
scan.rate-limit.records-per-second	限制Source每秒下發的最大記錄數。	否	LONG	無	適用於需要限制資料讀取情境，此限制在全量和增量階段都會生效。 Source的`numRecordsOutPerSecond`指標反映整個資料流每秒鐘輸出的記錄數，可以根據這個指標對此參數進行調整。在全量讀取階段，通常需要降低每個批次讀取資料的條數進行配合，可以減少`scan.incremental.snapshot.chunk.size`參數值。
include-binlog-meta.enable	是否在訊息中攜帶PolarDB-X Binlog的原始資訊，如GTID，Binlog位點等	否	Boolean	false	適用於原始binlog同步情境，比如替換原有canal同步鏈路。
polardbx.binlog.ignore.archive-events.enabled	是否忽略 PolarDB-X Binlog 中的歸檔事件（主要是 `DELETE` 事件）。	BOOLEAN	否	false
polardbx.binlog.ignore.query-events.enabled	是否忽略 PolarDB-X Binlog 中的ROWS_QUERY_LOG_EVENT事件。	BOOLEAN	否	false
polardbx.binlog.include.tables	僅訂閱這些表的Binlog日誌。多個表名之間用逗號（`,`）分隔。	STRING	否	無	說明此參數只在消費PolarDB-X Binlog增量階段有效，不影響快照階段的全量讀取。
polardbx.binlog.exclude.tables	不訂閱這些表的Binlog日誌。多個表名之間用逗號（`,`）分隔。	STRING	否	無	說明此參數只在消費PolarDB-X Binlog增量階段有效，不影響快照階段的全量讀取。