示例一：整庫同步

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

示例二：分庫合并同步

對于分庫合并同步的場景，需要利用正則表達式的庫名來匹配所要同步的多個分庫。使用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語句

實時計算Flink版支持使用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;

實時計算Flink版提供MySQL整庫同步到Kafka的能力，通過引入Kafka作為中間層，并使用CDAS整庫同步或CTAS整表同步到Kafka來解決，具體操作請參見MySQL整庫同步Kafka。