示例一：?jiǎn)伪硗?/b>

通常，CTAS都會(huì)配合數(shù)據(jù)源的Catalog和目標(biāo)的Catalog一起使用，例如MySQL Catalog和Hologres Catalog結(jié)合CTAS語(yǔ)法，來(lái)完成MySQL到Hologres的全量和增量數(shù)據(jù)同步。使用MySQL Catalog可以自動(dòng)解析源表的Schema及相應(yīng)的參數(shù)，而不用手動(dòng)編寫DDL 。

假設(shè)已在工作空間中注冊(cè)了名為holo的Hologres Catalog和名為mysql的MySQL Catalog。將MySQL中的web_sales表同步到Hologres中，代碼示例如下。

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales  
WITH ('jdbcWriteBatchSize' = '1024')   -- 可選，指定結(jié)果表的參數(shù)。
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- 指定mysql-cdc源表的額外參數(shù)。

示例二：分庫(kù)分表合并同步

對(duì)于分庫(kù)分表合并同步的場(chǎng)景，您可以結(jié)合MySQL Catalog，利用正則表達(dá)式的表名和庫(kù)名來(lái)匹配所要同步的多張表。使用CTAS可以將這多張分庫(kù)分表合并到一張Hologres表中，庫(kù)名和表名會(huì)作為額外的兩個(gè)字段寫入到該表中，為保證主鍵唯一性，庫(kù)名、表名和原主鍵一起作為該Hologres表的新聯(lián)合主鍵。

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`  
/*+ OPTIONS('server-id'='8001-8004') */;

其合并的效果如下圖所示。如果在user02表中新增一列age，并插入一條數(shù)據(jù)。此時(shí)雖然多張分表的Schema并不一致，但是user02表后續(xù)的數(shù)據(jù)和Schema變更都能實(shí)時(shí)地自動(dòng)同步到下游表中。

ALTER TABLE `user02` ADD COLUMN `age` INT;
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30);

示例三：自定義計(jì)算列同步

本示例以u(píng)ser分庫(kù)分表合并同步作為基礎(chǔ)，介紹在分庫(kù)分表合并的過(guò)程中，如何進(jìn)行一些轉(zhuǎn)換計(jì)算。

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3  AFTER `id`
);

新增計(jì)算列同步的效果如下圖所示。

示例四：多個(gè)CTAS語(yǔ)句作為一個(gè)作業(yè)提交

實(shí)時(shí)計(jì)算Flink版支持使用STATEMENT SET語(yǔ)法將多個(gè)CTAS語(yǔ)句作為一個(gè)作業(yè)一起提交，并且可以對(duì)Source進(jìn)行優(yōu)化，復(fù)用一個(gè)Source節(jié)點(diǎn)讀取多業(yè)務(wù)表的數(shù)據(jù)。這對(duì)于MySQL CDC數(shù)據(jù)源場(chǎng)景尤為適用，因?yàn)檫@可以減少server-id的使用，減少對(duì)數(shù)據(jù)庫(kù)的連接數(shù)和讀取壓力。

例如示例一同步了web_sales表，示例二同步了user分庫(kù)分表，您可以使用STATEMENT SET語(yǔ)法將它們作為一個(gè)作業(yè)提交。

USE CATALOG holo;

BEGIN STATEMENT SET;

-- 同步web_sales表。
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

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

END;

示例五：多個(gè)CTAS語(yǔ)句將同一張數(shù)據(jù)源表同步到不同的結(jié)果表

4.0.16以上版本中，在不添加計(jì)算列時(shí)，可以將同一張數(shù)據(jù)源表同步到不同的結(jié)果表。

USE CATALOG `holo`;

BEGIN STATEMENT SET;

-- 通過(guò)CTAS語(yǔ)句同步MySQL的user表到Holo數(shù)倉(cāng)database1的user表中
CREATE TABLE IF NOT EXISTS `database1`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

-- 通過(guò)CTAS語(yǔ)句同步MySQL的user表到Holo數(shù)倉(cāng)database2的user表中
CREATE TABLE IF NOT EXISTS `database2`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

如果結(jié)果表需要添加計(jì)算列，則應(yīng)按照如下方式進(jìn)行同步：

-- 基于源表user創(chuàng)建臨時(shí)表user_with_changed_id，支持定義計(jì)算列，例如這里的computed_id是基于源表的id計(jì)算獲得。
CREATE TEMPORARY TABLE `user_with_changed_id` (
  `computed_id` AS `id` + 1000
) LIKE `mysql`.`tpcds`.`user`;

-- 基于源表user創(chuàng)建臨時(shí)表user_with_changed_age，支持定義計(jì)算列，例如這里的computed_age是基于源表的age計(jì)算獲得。
CREATE TEMPORARY TABLE `user_with_changed_age` (
  `computed_age` AS `age` + 1
) LIKE `mysql`.`tpcds`.`user`;

BEGIN STATEMENT SET;

-- 通過(guò)CTAS語(yǔ)句同步MySQL的user表到Holo數(shù)倉(cāng)的user_with_changed_id表中，表中會(huì)包含通過(guò)計(jì)算獲得的id，即computed_id列。 
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
AS TABLE `user_with_changed_id`
/*+ OPTIONS('server-id'='8001-8004') */;

-- 通過(guò)CTAS語(yǔ)句同步MySQL的user表到Holo數(shù)倉(cāng)的user_with_changed_age表中，表中會(huì)包含通過(guò)計(jì)算獲得的age，即computed_age列。 
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
AS TABLE `user_with_changed_age`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

示例六：多個(gè)CTAS語(yǔ)句時(shí)，新增CTAS語(yǔ)句加入數(shù)據(jù)同步作業(yè)

使用VVR 8.0.1及以上版本時(shí)，多個(gè)CTAS語(yǔ)句的作業(yè)啟動(dòng)后，如果新增CTAS語(yǔ)句，支持從作業(yè)快照重啟，從而捕獲到新的表，對(duì)新增表進(jìn)行數(shù)據(jù)同步。

1. SQL作業(yè)開(kāi)發(fā)時(shí)需要增加以下語(yǔ)句，開(kāi)啟新增表讀取功能。

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

2. 當(dāng)需要新增CTAS語(yǔ)句時(shí)，在作業(yè)運(yùn)維頁(yè)面停止作業(yè)并勾選停止前創(chuàng)建一次快照。

3. 在SQL開(kāi)發(fā)中，增加CTAS語(yǔ)句，并重新部署這個(gè)SQL作業(yè)。

4. 在作業(yè)運(yùn)維頁(yè)面單擊目標(biāo)作業(yè)名稱，狀態(tài)集管理頁(yè)簽，單擊歷史。

5. 在作業(yè)快照列表中，找到停止作業(yè)時(shí)創(chuàng)建的快照。

6. 單擊目標(biāo)快照操作列，選擇更多 > 從該快照恢復(fù)作業(yè)。

7. 在作業(yè)啟動(dòng)配置對(duì)話框，配置作業(yè)啟動(dòng)信息，詳情請(qǐng)參見(jiàn)作業(yè)啟動(dòng)。

示例七：通過(guò)CTAS語(yǔ)句將MySQL數(shù)據(jù)源表同步到Hologres分區(qū)表

Hologres分區(qū)表建表時(shí)，如果Hologres表存在主鍵，則要求分區(qū)字段必須是主鍵中的字段。假設(shè)有一張MySQL表需要同步到Hologres，其建表語(yǔ)句如下。

CREATE TABLE orders (
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    city VARCHAR(100) NOT NULL
    order_date DATE,
    purchaser INTEGER,
    PRIMARY KEY(order_id, product_id)
);

當(dāng)使用CTAS同步數(shù)據(jù)源表到Hologres的分區(qū)表中時(shí)：

• 如果上游表的主鍵包含分區(qū)字段，例如Hologres表的分區(qū)字段是product_id，可以通過(guò)如下SQL實(shí)現(xiàn)。

  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
  PARTITIONED BY (product_id)
  AS TABLE `mysql`.`tpcds`.`orders`;

• 如果上游表的主鍵不包含分區(qū)字段，例如Hologres表的分區(qū)字段是city，創(chuàng)建Hologres表時(shí)會(huì)使用MySQL表中的主鍵，由于上游表的主鍵不包含分區(qū)字段，作業(yè)會(huì)出錯(cuò)。此時(shí)，您可以在CTAS中通過(guò)聲明主鍵的方式，重新指定目標(biāo)Hologres分區(qū)表的主鍵，使得任務(wù)正常運(yùn)行，示例如下。

  -- 可以通過(guò)如下SQL指定Hologres分區(qū)表的主鍵為order_id，product_id和city。
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`(
      CONSTRAINT `PK_order_id_city` PRIMARY KEY (`order_id`,`product_id`,`city`) NOT ENFORCED
  )
  PARTITIONED BY (city)
  AS TABLE `mysql`.`tpcds`.`orders`;

示例八：CTAS語(yǔ)句使用字段類型寬容模式同步數(shù)據(jù)到Hologres表

在CTAS場(chǎng)景中，可能需要調(diào)整已有字段數(shù)據(jù)類型的精度（例如，從VARCHAR(10)到VARCHAR(20)）。

• Flink計(jì)算引擎VVR 6.0.5-Flink 1.15以下版本，上游修改數(shù)據(jù)類型可能導(dǎo)致CTAS任務(wù)失敗，只能重建結(jié)果表。

• Flink計(jì)算引擎VVR 6.0.5-Flink 1.15及以上版本，在同步數(shù)據(jù)到Hologres表時(shí)，支持使用類型寬容模式。寬容模式應(yīng)該在首次啟動(dòng)CTAS作業(yè)時(shí)開(kāi)啟，如果在首次啟動(dòng)時(shí)未開(kāi)啟寬容模式，需要?jiǎng)h除下游表并且將作業(yè)無(wú)狀態(tài)重啟才能生效。

  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders` 
  WITH (
  'connector' = 'hologres', 
  'enableTypeNormalization' = 'true' -- 使用字段類型寬容模式。
  ) AS TABLE `mysql`.`tpcds`.`orders`;

  在上游發(fā)生數(shù)據(jù)類型修改事件時(shí)，只要所修改類型與原類型的歸一化類型相同，都視作修改成功。目前類型歸一化規(guī)則如下：

  • TINYINT、SMALLINT、INT和BIGINT歸一化為BIGINT。

  • CHAR、VARCHAR和STRING歸一化為STRING。

  • FLOAT和DOUBLE歸一化為DOUBLE。

  • 其他數(shù)據(jù)類型按照原本的類型映射規(guī)則創(chuàng)建，詳情參見(jiàn)類型映射。

  例如：

  • SMALLINT修改為INT，兩者的歸一化類型都是BIGINT，視為修改成功，CTAS作業(yè)正常運(yùn)行。

  • 從FLOAT改為BIGINT，兩者的歸一化類型分別為DOUBLE和BIGINT，屬于不兼容的情況，會(huì)拋出異常。

示例九：通過(guò)CTAS語(yǔ)句將MongoDB數(shù)據(jù)源表同步到Hologres表

實(shí)時(shí)計(jì)算Flink VVR 8.0.6及以上版本，CTAS語(yǔ)句支持同步MongoDB數(shù)據(jù)源表，能夠在實(shí)時(shí)同步MongoDB數(shù)據(jù)的同時(shí)將上游表結(jié)構(gòu)變更同步到下游表?？梢耘浜螹ongoDB Catalog使用，無(wú)需手動(dòng)定義Schema，MongoDB Catalog詳情可參考管理MongoDB Catalog。

這里以使用CTAS語(yǔ)句同步MongoDB數(shù)據(jù)源表數(shù)據(jù)到Hologres表為例：

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table1`
AS TABLE `mongodb`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table2`
AS TABLE `mongodb`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

END;

當(dāng)使用同一個(gè)作業(yè)同步多個(gè)MongoDB集合時(shí)，需要滿足以下條件：

• 每張表關(guān)于MongoDB的配置必須完全相同，包括hosts、scheme、username、password、connectionOptions。

• 每張表的scan.startup.mode配置必須完全相同。

示例十：MySQL整庫(kù)同步Kafka

在實(shí)際使用中，同一張MySQL表可能被多個(gè)作業(yè)依賴，當(dāng)多個(gè)任務(wù)使用同一張MySQL表做處理時(shí)，MySQL數(shù)據(jù)庫(kù)會(huì)啟動(dòng)多個(gè)連接，對(duì)MySQL服務(wù)器和網(wǎng)絡(luò)造成很大的壓力。為了緩解對(duì)上游MySQL數(shù)據(jù)庫(kù)的壓力，實(shí)時(shí)計(jì)算Flink版提供MySQL整庫(kù)同步到Kafka的能力，通過(guò)引入Kafka作為中間層，并使用CDAS整庫(kù)同步或CTAS整表同步到Kafka來(lái)解決。具體操作請(qǐng)參見(jiàn)MySQL整庫(kù)同步Kafka。