Realtime Compute for Apache Flink：CREATE TABLE AS（CTAS）語句

樣本一：單表同步

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

假設已在工作空間中註冊了名為holo的Hologres Catalog和名為mysql的MySQL Catalog。將MySQL中的web_sales表同步到Hologres中，程式碼範例如下。

USE CATALOG holo;

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

樣本二：分庫分表合并同步

對於分庫分表合并同步的情境，您可以結合MySQL Catalog，利用Regex的表名和庫名來匹配所要同步的多張表。使用CTAS可以將這多張分庫分表合并到一張Hologres表中，庫名和表名會作為額外的兩個欄位寫入到該表中，為保證主鍵唯一性，庫名、表名和原主鍵一起作為該Hologres表的新聯合主鍵。

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，並插入一條資料。此時雖然多張分表的Schema並不一致，但是user02表後續的資料和Schema變更都能即時地自動同步到下遊表中。

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

樣本三：自訂計算列同步

本樣本以user分庫分表合并同步作為基礎，介紹在分庫分表合并的過程中，如何進行一些轉換計算。

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`
);

新增計算資料行同步的效果如下圖所示。

樣本四：多個CTAS語句作為一個作業提交

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

例如樣本一同步了web_sales表，樣本二同步了user分庫分表，您可以使用STATEMENT SET文法將它們作為一個作業提交。

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分庫分表。
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

樣本五：多個CTAS語句將同一張資料來源表同步到不同的結果表

4.0.16以上版本中，在不添加計算資料行時，可以將同一張資料來源表同步到不同的結果表。

USE CATALOG `holo`;

BEGIN STATEMENT SET;

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

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

END;

如果結果表需要添加計算資料行，則應按照如下方式進行同步：

-- 基於源表user建立暫存資料表user_with_changed_id，支援定義計算資料行，例如這裡的computed_id是基於源表的id計算獲得。
CREATE TEMPORARY TABLE `user_with_changed_id` (
  `computed_id` AS `id` + 1000
) LIKE `mysql`.`tpcds`.`user`;

-- 基於源表user建立暫存資料表user_with_changed_age，支援定義計算資料行，例如這裡的computed_age是基於源表的age計算獲得。
CREATE TEMPORARY TABLE `user_with_changed_age` (
  `computed_age` AS `age` + 1
) LIKE `mysql`.`tpcds`.`user`;

BEGIN STATEMENT SET;

-- 通過CTAS語句同步MySQL的user表到Holo數倉的user_with_changed_id表中，表中會包含通過計算獲得的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') */;

-- 通過CTAS語句同步MySQL的user表到Holo數倉的user_with_changed_age表中，表中會包含通過計算獲得的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;

樣本六：多個CTAS語句時，新增CTAS語句加入資料同步作業

使用VVR 8.0.1及以上版本時，多個CTAS語句的作業啟動後，如果新增CTAS語句，支援從作業快照重啟，從而捕獲到新的表，對新增表進行資料同步。

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

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

2. 當需要新增CTAS語句時，在作業營運頁面停止作業並勾選停止前建立一次快照。

3. 在SQL開發中，增加CTAS語句，並重新部署這個SQL作業。

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

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

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

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

樣本七：通過CTAS語句將MySQL資料來源表同步到Hologres分區表

Hologres分區表建表時，如果Hologres表存在主鍵，則要求分區欄位必須是主鍵中的欄位。假設有一張MySQL表需要同步到Hologres，其建表語句如下。

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)
);

當使用CTAS同步資料來源表到Hologres的分區表中時：

• 如果上遊表的主鍵包含分區欄位，例如Hologres表的分區欄位是product_id，可以通過如下SQL實現。

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

• 如果上遊表的主鍵不包含分區欄位，例如Hologres表的分區欄位是city，建立Hologres表時會使用MySQL表中的主鍵，由於上遊表的主鍵不包含分區欄位，作業會出錯。此時，您可以在CTAS中通過聲明主鍵的方式，重新指定目標Hologres分區表的主鍵，使得任務正常運行，樣本如下。

  -- 可以通過如下SQL指定Hologres分區表的主鍵為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語句使用欄位類型寬容模式同步資料到Hologres表

在CTAS情境中，可能需要調整已有欄位資料類型的精度（例如，從VARCHAR(10)到VARCHAR(20)）。

• Flink計算引擎VVR 6.0.5-Flink 1.15以下版本，上遊修改資料類型可能導致CTAS任務失敗，只能重建結果表。

• Flink計算引擎VVR 6.0.5-Flink 1.15及以上版本，在同步資料到Hologres表時，支援使用類型寬容模式。寬容模式應該在初次開機CTAS作業時開啟，如果在初次開機時未開啟寬容模式，需要刪除下遊表並且將作業無狀態重啟才會生效。

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

  在上遊發生資料類型修改事件時，只要所修改類型與原類型的歸一化類型相同，都視作修改成功。目前類型歸一化規則如下：

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

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

  • FLOAT和DOUBLE歸一化為DOUBLE。

  • 其他資料類型按照原本的類型映射規則建立，詳情參見類型映射。

  例如：

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

  • 從FLOAT改為BIGINT，兩者的歸一化類型分別為DOUBLE和BIGINT，屬於不相容的情況，會拋出異常。

樣本九：通過CTAS語句將MongoDB資料來源表同步到Hologres表

Realtime ComputeFlink VVR 8.0.6及以上版本，CTAS語句支援同步MongoDB資料來源表，能夠在即時同步MongoDB資料的同時將上遊表結構變更同步到下遊表。可以配合MongoDB Catalog使用，無需手動定義Schema，MongoDB Catalog詳情可參考管理MongoDB Catalog。

這裡以使用CTAS語句同步MongoDB資料來源表資料到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;

當使用同一個作業同步多個MongoDB集合時，需要滿足以下條件：

• 每張表關於MongoDB的配置必須完全相同，包括hosts、scheme、username、password、connectionOptions。

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

樣本十：MySQL整庫同步Kafka

在實際使用中，同一張MySQL表可能被多個作業依賴，當多個任務使用同一張MySQL表做處理時，MySQL資料庫會啟動多個串連，對MySQL伺服器和網路造成很大的壓力。為了緩解對上遊MySQL資料庫的壓力，Realtime ComputeFlink版提供MySQL整庫同步到Kafka的能力，通過引入Kafka作為中介層，並使用CDAS整庫同步或CTAS整表同步到Kafka來解決。具體操作請參見MySQL整庫同步Kafka。