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

樣本一：整庫同步

CDAS通常會配合資料來源的Catalog和目標的Catalog一起使用。例如，MySQL Catalog和Hologres Catalog結合CDAS文法，完成MySQL到Hologres的全量和增量資料同步。使用MySQL Catalog可以自動解析源表的Schema及相應的參數，而不用手動編寫DDL。

假設已在工作空間中註冊了名為holo的Hologres Catalog和名為mysql的MySQL Catalog，MySQL中有一個名為tpcds的庫。您可以使用以下語句將tpcds庫下的24張表全部同步到Hologres中，包括未來的資料變更和表結構變更，無需提前在Hologres中建立表。

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源表的額外參數。

樣本二：分庫合并同步

對於分庫合并同步的情境，需要利用Regex的庫名來匹配所要同步的多個分庫。使用CDAS可以將上遊多個分庫下相同表名的資料合併同步到Hologres目標庫對應表名的同一張表中，庫名和表名會作為額外的兩個欄位寫入到每張結果表中。為保證主鍵唯一性，庫名、表名和原主鍵一起作為對應Hologres表的新聯合主鍵。

假設MySQL執行個體中有order_db01~order_db99多個分庫，每個分庫下都有order、order_detail等多張表。您可以使用以下語句將99個分庫下的order、order_detail等表全部同步到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源表的額外參數。

樣本三：源庫新增表加入資料同步

使用VVR 8.0.6及以上版本時，CDAS作業啟動後，如果源庫新增表，支援從作業快照重啟，從而捕獲到新的表，對新增表進行資料同步。

1. SQL作業開發時需要增加以下語句，開啟CDAS新增表讀取功能。

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

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

3. 在SQL開發中，重新部署這個SQL作業。

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

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

6. 單擊目標快照操作列，選擇更多 > 從該快照恢複作業。

7. 在作業啟動配置對話方塊，配置作業啟動資訊，詳情請參見作業啟動。

樣本四：多CDAS&CTAS語句

Realtime ComputeFlink版支援使用STATEMENT SET文法將多個CTAS語句作為一個作業一起提交，並且可以對Source進行最佳化，複用一個Source節點讀取多業務表的資料。這對於MySQL CDC資料來源情境尤為適用，因為這可以減少server-id的使用，減少對資料庫的串連數和讀取壓力。

假設MySQL執行個體中有tpcds、tpch、user_db01~user_db99（分庫分表）多個庫。您可以通過組合多條CDAS和CTAS語句，將MySQL執行個體下的所有庫和表都同步到Hologres，只需一個Flink作業便能完成所有表的同步，只需一個Source便能讀取所有表的資料，程式碼範例如下。

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

在使用多個CDAS語句整庫同步到Kafka時，由於不同的資料庫中可能存在相同的表，為了防止topic衝突，需要使用cdas.topic.pattern配置。cdas.topic.pattern定義了建立topic的名稱的格式，其中可通過{table-name}預留位置來替換為表名。如：當設定'cdas.topic.pattern'='db1-{table-name}'，對於上遊表名為table1的表，在Kafka中對應的topic名稱為db1-table1。

假設MySQL執行個體中有tpcds、tpch多個庫。您可以通過如下方式將MySQL執行個體下的所有庫和表都同步到Kafka，避免topic衝突，程式碼範例如下。

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。