兼容PolarDB PostgreSQL版(兼容Oracle)的Flink CDC連接器(簡稱PolarDBO Flink CDC)可用于依次讀取PolarDB PostgreSQL版(兼容Oracle)數據庫全量快照數據和變更數據,具體功能及用法請參考社區Postgres CDC。
由于PolarDB PostgreSQL版(兼容Oracle)與社區PoatgreSQL僅在少量數據類型和內置對象處理存在差異,本文為您介紹如何基于社區Postgres CDC,通過少量代碼適配打包出支持PolarDB PostgreSQL版(兼容Oracle)的PolarDBO Flink CDC連接器。
PolarDB PostgreSQL版(兼容Oracle)的DATE類型是64位,而社區PostgreSQL的DATE類型為32位。因此,在PolarDBO Flink CDC中會對DATA類型數據的處理進行適配。
打包PolarDBO Flink CDC連接器
PolarDBO Flink CDC連接器基于社區Postgres CDC適配開發,無論是您自行打包,還是使用本文中提供的JAR包,PolarDBO Flink CDC連接器都不提供SLA保障。
操作前提
確定Flink-CDC版本
如果您使用的是阿里云實時計算 Flink 版,需要確認與對應Ververica Runtime(簡稱VVR)版本兼容的社區Flink-CDC版本,具體可以參考CDC與VVR版本對應關系。
說明Flink-CDC代碼倉庫請參考Flink-CDC。
確定Debezium版本
在對應版本的Flink-CDC的
pom.xml
中通過查找關鍵字debezium.version
確定Debezium版本。說明Debezium代碼倉庫請參考Debezium。
確定PgJDBC版本
在對應版本的Postgres-CDC的
pom.xml
中通過查找關鍵字org.postgresql
確定PgJDBC版本。說明release-3.0以下版本文件路徑為:
flink-connector-postgres-cdc/pom.xml
。release-3.0及以上版本文件路徑為:
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml
。PgJDBC代碼倉庫請參考PgJDBC。
操作步驟
release-3.1打包
社區Flink-CDC release-3.1版本兼容阿里云實時計算 Flink 版的vvr-8.0.x-flink-1.17。
打包對應版本的PolarDBO Flink CDC連接器步驟如下:
克隆對應版本的Flink-CDC、Debezium和PgJDBC的代碼文件。
git clone -b release-3.1 --depth=1 https://github.com/apache/flink-cdc.git git clone -b REL42.5.1 --depth=1 https://github.com/pgjdbc/pgjdbc.git git clone -b v1.9.8.Final --depth=1 https://github.com/debezium/debezium.git
復制Debezium和PgJDBC部分文件到Flink-CDC中。
mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 mkdir -p flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 cp pgjdbc/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc cp debezium/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql
應用適配PolarDB PostgreSQL版(兼容Oracle)的patch文件。
git apply release-3.1_support_polardbo.patch
說明以上使用的PolarDBO Flink CDC兼容patch文件:release-3.1_support_polardbo.patch。
使用Maven打包PolarDBO Flink CDC連接器。
mvn clean install -DskipTests -Dcheckstyle.skip=true -Dspotless.check.skip -Drat.skip=true # 打包完成后可以在flink-sql-connector-postgres-cdc的target目錄中獲取到jar包
按照以上流程基于JDK8打包出PolarDBO Flink CDC連接器的JAR包:flink-sql-connector-postgres-cdc-3.1-SNAPSHOT.jar。
release-2.3打包
社區Flink-CDC release-2.3版本兼容阿里云實時計算 Flink 版的vvr-4.0.15-flink-1.13 ~ vvr-6.0.2-flink-1.15。
打包對應版本的PolarDBO Flink CDC連接器步驟如下:
克隆對應版本的Flink-CDC、Debezium和PgJDBC的代碼文件。
git clone -b release-2.3 --depth=1 https://github.com/apache/flink-cdc.git git clone -b REL42.2.26 --depth=1 https://github.com/pgjdbc/pgjdbc.git git clone -b v1.6.4.Final --depth=1 https://github.com/debezium/debezium.git
復制Debezium和PgJDBC部分文件到Flink-CDC中。
mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc mkdir -p flink-cdc/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/ConnectionFactoryImpl.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 cp pgjdbc/pgjdbc/src/main/java/org/postgresql/core/v3/QueryExecutorImpl.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/core/v3 cp pgjdbc/pgjdbc/src/main/java/org/postgresql/jdbc/PgDatabaseMetaData.java flink-cdc/flink-connector-postgres-cdc/src/main/java/org/postgresql/jdbc cp debezium/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/TypeRegistry.java flink-cdc/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql
應用適配PolarDB PostgreSQL版(兼容Oracle)的patch文件。
git apply release-2.3_support_polardbo.patch
說明以上使用的PolarDBO Flink CDC兼容patch文件:release-2.3_support_polardbo.patch。
使用Maven打包PolarDBO Flink CDC連接器。
mvn clean install -DskipTests -Dcheckstyle.skip=true -Dspotless.check.skip -Drat.skip=true # 打包完成后可以在flink-sql-connector-postgres-cdc的target目錄中獲取到jar包
按照以上流程基于JDK8打包出PolarDBO Flink CDC連接器的JAR包:flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar。
使用說明
PolarDBO Flink CDC連接器通過PolarDB PostgreSQL版(兼容Oracle)數據庫的邏輯復制讀取CDC變更流數據,需要滿足以下條件:
wal_level參數的值需設置為logical,即在預寫式日志WAL(Write-ahead logging)中增加支持邏輯復制所需的信息。
說明您可以通過控制臺設置wal_level參數,詳細操作請參考設置集群參數。修改該參數后集群將會重啟,請在修改參數前做好業務安排,謹慎操作。
執行
ALTER TABLE schema.table REPLICA IDENTITY FULL;
命令設置訂閱表的REPLICA IDENTITY
為FULL
(發出的插入和更新操作事件包含表中所有列的舊值),以保障該表數據同步的一致性。說明REPLICA IDENTITY是PostgreSQL特有的表級設置,決定了邏輯解碼插件在發生(INSERT)和更新(UPDATE)事件時,是否包含涉及的表列的舊值。REPLICA IDENTITY取值含義詳情,請參見REPLICA IDENTITY。
設置訂閱表的
REPLICA IDENTITY
為FULL
時可能需要鎖表,可能影響業務,請在修改參數前做好業務安排。您可以通過以下命令查看當前配置是否為FULL
:SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';
需要確保max_wal_senders和max_replication_slots的參數值均大于當前數據庫復制槽已使用數與Flink作業所需要的slot數量。
確保使用的是高權限賬號或者同時擁有LOGIN和REPLICATION權限,并且具有訂閱表的SELECT權限用于全量數據查詢。
只能連接PolarDB集群的主地址,集群地址不支持邏輯復制。
PolarDBO Flink CDC連接器與Postgres CDC區別
PolarDBO Flink CDC連接器基于Postgres CDC打包,具體語法和參數可以參考Postgres CDC。但存在以下主要區別:
WITH的connector參數需要設置為固定值:
polardbo-cdc
。PolarDBO Flink CDC同時兼容PolarDB PostgreSQL版各版本、PolarDB PostgreSQL版(兼容Oracle) 1.0和PolarDB PostgreSQL版(兼容Oracle) 2.0版本。
說明如果您使用的是PolarDB PostgreSQL版,推薦您直接使用社區Postgres CDC。
PolarDB PostgreSQL版(兼容Oracle) 1.0、PolarDB PostgreSQL版(兼容Oracle) 2.0中的
DATE
類型的列,Flink SQL中的source和sink表對應類型必須指定為timestamp
。建議將
decoding.plugin.name
參數設置為pgoutput
,否則非UTF-8編碼的數據庫可能會發生增量解析亂碼,詳細介紹請參考社區文檔。
類型映射
PolarDB PostgreSQL和Flink字段類型映射,除DATE類型外,其他字段類型和社區PostgreSQL完全相同,具體映射關系如下:
PolarDB PostgreSQL字段類型 | Flink字段類型 |
SMALLINT | SMALLINT |
INT2 | |
SMALLSERIAL | |
SERIAL2 | |
INTEGER | INT |
SERIAL | |
BIGINT | BIGINT |
BIGSERIAL | |
REAL | FLOAT |
FLOAT4 | |
FLOAT8 | DOUBLE |
DOUBLE PRECISION | |
NUMERIC(p, s) | DECIMAL(p, s) |
DECIMAL(p, s) | |
BOOLEAN | BOOLEAN |
DATE |
|
TIME [(p)] [WITHOUT TIMEZONE] | TIME [(p)] [WITHOUT TIMEZONE] |
TIMESTAMP [(p)] [WITHOUT TIMEZONE] | TIMESTAMP [(p)] [WITHOUT TIMEZONE] |
CHAR(n) | STRING |
CHARACTER(n) | |
VARCHAR(n) | |
CHARACTER VARYING(n) | |
TEXT | |
BYTEA | BYTES |
示例
以下示例用于說明,如何通過PolarDBO Flink CDC,將PolarDB PostgreSQL版(兼容Oracle) 2.0集群中flink_source庫的shipments表,同步到flink_sink庫的shipments_sink表中。
以下示例僅用于簡單驗證打包的PolarDBO Flink CDC能夠在PolarDB PostgreSQL版(兼容Oracle)上運行。正式使用時,為滿足您的業務需求,請參考社區Postgres CDC配置參數。
前提準備
PolarDB PostgreSQL版(兼容Oracle)準備
在PolarDB集群購買頁面,購買PolarDB PostgreSQL版(兼容Oracle) 2.0集群。
創建高權限賬戶,詳細操作請參考創建賬號。
獲取集群主地址,詳細操作請參考查看連接地址。如果PolarDB集群和實時計算 Flink 版在同一可用區,可直接使用私網地址,否則需要申請公網地址。將Flink實例地址添加到PolarDB集群白名單中,請參考設置集群白名單。
在控制臺創建源數據庫flink_source和目標數據庫flink_sink,詳細步驟請參考創建數據庫。
執行如下語句,在源數據庫flink_source中創建shipments表,并寫入數據。
CREATE TABLE public.shipments ( shipment_id INT, order_id INT, origin TEXT, destination TEXT, is_arrived BOOLEAN, order_time DATE, PRIMARY KEY (shipment_id) ); ALTER TABLE public.shipments REPLICA IDENTITY FULL; INSERT INTO public.shipments SELECT 1, 1, 'test1', 'test1', false, now();
執行如下語句,在目標數據庫flink_sink中創建shipments_sink表。
CREATE TABLE public.shipments_sink ( shipment_id INT, order_id INT, origin TEXT, destination TEXT, is_arrived BOOLEAN, order_time TIMESTAMP, PRIMARY KEY (shipment_id) );
實時計算 Flink 版準備
登錄實時計算控制臺,購買實時計算 Flink 版實例,詳細操作請參考開通實時計算Flink版。
說明建議實時計算 Flink 版的地域和專有網絡和PolarDB集群保持一致,連接地址可以直接使用PolarDB集群主地址的私網地址。
創建自定義連接器,上傳打包好的PolarDBO Flink CDC,Formats選擇debezium-json,詳細步驟請參考創建自定義連接器。
創建Flink作業
登錄實時計算控制臺,新建一個SQL作業草稿,請參考SQL作業開發。使用以下Flink SQL語句,修改PolarDB集群主地址,端口,賬號和密碼。
說明PolarDB PostgreSQL版(兼容Oracle)的DATE類型是64位,而Flink SQL以及大部分數據庫的DATE類型為32位。因此,源表中DATE類型的列,在Flink SQL的source和sink表中都必須要指定為TIMESTAMP類型。否則,作業會因為類型不匹配而報錯中斷,例如:
“java.time.DateTimeException: Invalid value for EpochDay (valid values -365243219162 - 365241780471):1720891573000”
。CREATE TEMPORARY TABLE shipments ( shipment_id INT, order_id INT, origin STRING, destination STRING, is_arrived BOOLEAN, order_time TIMESTAMP, PRIMARY KEY (shipment_id) NOT ENFORCED ) WITH ( 'connector' = 'polardbo-cdc', 'hostname' = '<yourHostname>', 'port' = '<yourPort>', 'username' = '<yourUserName>', 'password' = '<yourPassWord>', 'database-name' = 'flink_source', 'schema-name' = 'public', 'table-name' = 'shipments', 'decoding.plugin.name' = 'pgoutput', 'slot.name' = 'flink' ); CREATE TEMPORARY TABLE shipments_sink ( shipment_id INT, order_id INT, origin STRING, destination STRING, is_arrived BOOLEAN, order_time TIMESTAMP, PRIMARY KEY (shipment_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:postgresql://<yourHostname>:<yourPort>/flink_sink', 'table-name' = 'shipments_sink', 'username' = '<yourUserName>', 'password' = '<yourPassWord>' ); INSERT INTO shipments_sink SELECT * FROM shipments;
部署并啟動作業。
測試與驗證。
部署作業運行成功后,即狀態為運行中,shipments表中的數據已經同步到目標數據庫flink_sink的shipments_sink表。
SELECT * FROM public.shipments_sink;
返回結果如下:
shipment_id | order_id | origin | destination | is_arrived | order_time -------------+----------+--------+-------------+------------+--------------------- 1 | 1 | test1 | test1 | f | 2024-09-18 05:45:08 (1 row)
在源數據庫flink_source的shipments表上執行DML,新增修改也將實時同步。
INSERT INTO public.shipments SELECT 2, 2, 'test2', 'test2', false, now(); UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 1; DELETE FROM public.shipments WHERE shipment_id = 2; INSERT INTO public.shipments SELECT 3, 3, 'test3', 'test3', false, now(); UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 3;
shipments表中的數據已經同步更新到目標數據庫flink_sink的shipments_sink表。
SELECT * FROM public.shipments_sink;
返回結果如下:
shipment_id | order_id | origin | destination | is_arrived | order_time -------------+----------+--------+-------------+------------+--------------------- 1 | 1 | test1 | test1 | t | 2024-09-18 05:45:08 3 | 3 | test3 | test3 | t | 2024-09-18 07:33:23 (2 rows)