Realtime Compute for Apache Flink：CREATE DATABASE AS（CDAS）語句

資料同步

表結構變更同步

在即時同步整庫資料的同時，還支援將每張源表的表結構變更（加列等）即時同步到結果表中。CDAS的表結構變更同步策略與CTAS一致，詳情請參見表結構變更同步。

文法限制

• 不支援調試功能。

• 不支援MiniBatch配置。

  重要

  • 建立SQL作業前：請確保組態管理頁面的作業預設配置頁簽中的其他配置處刪除了MiniBatch配置。

  • 建立SQL作業後：具體解決方案可參見報錯：Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax。

上下遊儲存相容性

CDAS支援的上下遊儲存列表如下，您可以從下表的源表和結果表中各選一個進行組合。

整庫同步

同步情境：將MySQL中的tpcds庫下的所有表同步到Hologres。

前提條件：已在工作空間中註冊以下Catalog。

• Hologres Catalog：名稱為holo。

• MySQL Catalog：名稱為mysql。

程式碼範例：

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds  -- 在hologres中建立holo_tpcds庫。
WITH ('sink.parallelism' = '4') -- 可選，指定目標庫的參數，每個holo sink預設使用4並發。
AS DATABASE mysql.tpcds INCLUDING ALL TABLES  -- 同步mysql中tpcds庫下所有表。
/*+ OPTIONS('server-id'='8001-8004') */ ; -- 可選，指定mysql-cdc源表的額外參數。

分庫合并同步

同步情境：MySQL執行個體中有order_db01~order_db99多個分庫，每個分庫下都有order、order_detail等多張表。使用CDAS將MySQL的多個分庫下的所有表全部同步到Hologres中，包括未來的資料變更和表結構變更。

同步方案：利用Regex的庫名（`order_db[0-9]+`）來匹配所要同步的多個分庫（order_db01~order_db99）。其中庫名和表名會作為額外的兩個欄位寫入到每張結果表中。為保證主鍵唯一性，庫名、表名和原主鍵一起作為對應Hologres表的新聯合主鍵。

程式碼範例及合并效果：

使用CDAS可以將上遊多個分庫下相同表名的資料合併同步到Hologres目標庫對應表名的同一張表中，無需提前在Hologres中建立表。

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order--在Hologres中建立holo_order庫，包括mysql中order分庫的所有表。
WITH('sink.parallelism'='4')        --可選，指定目標庫的參數，每個HologresSink預設並發為4。
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES --同步mysql中order_db分庫下所有表。
/*+OPTIONS('server-id'='8001-8004')*/;  --可選，指定mysql-cdc源表的額外參數。

源庫新增表同步

同步情境：CDAS作業啟動後，需要對源庫新增表進行資料同步。

同步方案：在SQL作業中開啟新增表讀取功能，從作業快照重啟，從而捕獲到新的表，對新增表進行資料同步。

使用限制：VVR 8.0.6及以上版本支援源庫新增表同步功能，啟用該功能需確保源表啟動模式為initial。

操作步驟：

1. 當出現新增的表需要同步時，在作業營運頁面停止作業並勾選停止前建立一次快照。

2. 在SQL作業開發中開啟新增表讀取功能，然後重新部署作業。

   1. 在SQL作業中增加以下語句，開啟CDAS新增表讀取功能。

      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

   2. 單擊部署。

3. 從快照恢複作業。

   1. 在作業營運頁面單擊目標作業名稱，狀態集管理頁簽，單擊歷史。

   2. 在作業快照列表中，找到停止作業時建立的快照。

   3. 單擊目標快照操作列，選擇更多 > 從該快照恢複作業完成作業啟動。詳情請參見作業啟動。

多CDAS&CTAS語句

同步情境：通過一個作業將MySQL執行個體中tpcds、tpch、user_db01~user_db99（分庫分表）多個庫同步到Hologres。

同步方案：使用STATEMENT SET文法組合多條CDAS和CTAS語句作為一個作業提交。該方案可以複用一個Source節點讀取所需表的資料，在MySQL CDC資料來源情境可以減少server-id的使用及資料庫的串連數與讀取壓力。

程式碼範例：

USE CATALOG holo;

BEGIN STATEMENT SET;

-- 同步user分庫分表。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- 同步TPCDS庫。
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;
-- 同步TPCH庫。
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

多CDAS語句整庫同步到Kafka

同步情境：將MySQL執行個體中tpcds、tpch多個庫和表都同步到Kafka。

同步方案：在使用多個CDAS語句整庫同步到Kafka時，由於不同的資料庫中可能存在相同的表，為了防止topic衝突，需要使用cdas.topic.pattern配置。

cdas.topic.pattern定義了建立topic的名稱的格式，其中可通過{table-name}預留位置來替換為表名。例如：當設定'cdas.topic.pattern'='dbname-{table-name}'，對於上遊表名為table1的表，在Kafka中對應的topic名稱為dbname-table1。

程式碼範例：

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- 同步TPCDS庫。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

-- 同步TPCH庫。
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

Realtime ComputeFlink版提供MySQL整庫同步到Kafka的能力，通過引入Kafka作為中介層，並使用CDAS整庫同步或CTAS整表同步到Kafka來解決，具體操作請參見MySQL整庫同步Kafka。

常見問題

作業運行異常

• 作業啟動不起來，應該如何排查？

• 作業運行後出現重啟，應該如何排查呢？

• 請求逾時問題：報錯：akka.pattern.AskTimeoutException

作業效能問題

• 如何排查作業反壓問題？

• 全表讀取階段效率低且存在反壓，如何解決？

• 如何排查作業延遲過高的問題？

• 消費上遊資料速度不穩定，該如何排查？

資料同步問題

• 為什麼主鍵為bigint unsigned的MySQL表註冊Flink Catalog，主鍵會變為decimal？但是使用CTAS同步到Hologres後，主鍵又變為text？

• 使用MiniBatch配置報錯：報錯：Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax

• 為什麼作業沒有輸出？

• 如何定位Flink無法讀取來源資料的問題？

• 如何定位Flink無法將資料寫入到結果表的問題？

• 作業資料不準確，該如何排查？