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

資料同步

表結構變更同步

通過CTAS語句，在即時同步資料的同時，還能將源表Schema的變更同步到結果表中。Schema變更包括初始的表建立以及未來的表變更。

• 支援同步的Schema變更

  Schema變更	說明
  添加可空列	自動在結果表Schema末尾添加對應的列並同步資料。新增的列會預設設定為可空列，變更前該列的資料自動化佈建為NULL值。
  添加非空列	自動在結果表Schema末尾添加對應的列並同步資料。
  刪除可空列	不會直接在結果表中刪除該列，而是將該列資料自動化佈建為NULL值。
  重新命名列	等同於添加新列並刪除舊列，即在結果表Schema末尾添加重新命名後的列，並將重新命名前的列資料自動化佈建為NULL值。 說明 例如，如果col_a被重新命名為col_b，則會在結果表末尾添加col_b，並自動將col_a的資料填充為NULL值。
  列類型變更	下遊系統支援列類型變更：目前只有Paimon支援處理列類型變更。CTAS支援普通列的類型變更，例如，從INT類型變更到BIGINT類型。 此類變更依賴於下遊Sink支援的列類型變更規則，請參考對應結果表文檔擷取其支援的列類型變更規則。 下遊系統不支援列類型變更：目前只有Hologres支援寬容模式處理列類型變更，即CTAS作業啟動時建立類型更加寬泛的下遊表，通過下遊Sink對列類型變更的相容性實現列類型變更支援，具體可參見樣本：寬容模式同步資料。 重要 寬容模式應該在初次開機CTAS作業時開啟，如果在初次開機時未開啟寬容模式，需要刪除下遊表並且將作業無狀態重啟才會生效。

  重要

  CTAS是對比前後兩條資料的Schema差異，不會去識別具體的DDL類型。

  • 刪除了某列後重新添加該列，且期間無資料變化，那麼CTAS會認為沒有發生結構變更。

  • 添加新列後且有資料變化時，CTAS才會感知到結構變更，然後同步結構變更到結果表。

• 不支援同步的Schema變更

  • 不支援主鍵或索引等約束的變更。

  • 不支援非空列的刪除。

  • 不支援從NOT NULL轉為NULLABLE的變更。

  重要

  如果遇到以上不支援的Schema變更，需要您手動刪除下遊結果表，重新啟動CTAS作業，即重新建立結果表並重新同步歷史資料。

單表同步

同步情境：將MySQL中的web_sales表同步到Hologres。

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

• Hologres Catalog：名稱為holo。

• MySQL Catalog：名稱為mysql。

程式碼範例：

CTAS通常會配合資料來源的Catalog和目標的Catalog一起使用，其中源Catalog可以自動解析源表的Schema及參數（無需手動編寫DDL），最終完成源表到目標表的全量和增量資料同步。

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- 沒有指定資料庫，則同步到預設資料庫的web_sales表。
WITH ('jdbcWriteBatchSize' = '1024')   -- 可選，指定結果表的參數。
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- 指定mysql-cdc源表的額外參數。

分庫分表合并同步

同步情境：使用CTAS將MySQL多張分庫分表合并到一張Hologres表中。

同步方案：結合MySQL Catalog，利用Regex的庫名和表名匹配要同步的多張表。其中庫名和表名會作為額外的兩個欄位寫入到結果表中，為保證主鍵唯一性，庫名、表名和原主鍵會一起作為該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') */;

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

自訂計算列同步

同步情境：在MySQL分庫分表合并過程中新增自訂計算列同步到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') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3  AFTER `id`
);

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

同步情境：將MySQL中的web_sales表、user分庫分表作為一個作業同步到Hologres。

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

程式碼範例：

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語句：同一源表同步到多結果表

• 結果表不添加計算資料行

  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語句的作業啟動後，需要新增CTAS語句對新增表進行資料同步。

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

使用限制：

• VVR 8.0.1及以上版本支援新增表功能。

• 使用CDC源表同步時，僅支援源表啟動模式為initial的作業使用新增表功能。

• 新增CTAS語句中新增的源表配置要和原有的源表配置完全一致，確保Source能夠複用。

• 新增CTAS語句前後，作業配置參數不能變更（例如更改啟動模式等）。

操作步驟：

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

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

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

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

   2. 在SQL作業中增加CTAS語句，最終完整的程式碼範例如下。

      -- 開啟新增表讀取功能。
      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';
      
      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') */;
      
      -- 同步product表。（新增表）
      CREATE TABLE IF NOT EXISTS product
      AS TABLE mysql.tpcds.product
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      END;

   3. 單擊部署。

3. 從快照恢複作業。

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

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

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

同步到Hologres分區表

同步情境：通過CTAS語句將MySQL源表同步到Hologres分區表。

Hologres分區表規則：在Hologres中，如果目標表定義了主鍵，則分區欄位必須包含在主鍵列中。

程式碼範例：

MySQL源表的建表語句如下：

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會自動驗證分區欄位是否為主鍵的一部分。

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

• 源表主鍵不包含分區欄位：在CTAS語句中重新聲明目標表的主鍵。

  如果Hologres表的分區欄位（如city）不在上遊表主鍵中，則直接同步處理會導致作業失敗。您需要在CTAS語句中重新聲明目標表的主鍵，確保分區欄位成為主鍵的一部分。

  -- 可以通過如下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表時，需要支援調整已有欄位資料類型的精度（如從VARCHAR(10)修改為VARCHAR(20)）或者修改資料類型（如從SMALLINT修改為INT）的情境。

同步方案：使用Hologres欄位類型寬容模式同步資料。寬容模式應該在CTAS作業初次開機時開啟，若未開啟，需要刪除下遊表並且將作業無狀態重啟才會生效。

類型歸一化規則：

上遊資料類型修改後，若修改類型與原類型的歸一化類型相同，則視為類型修改成功，CTAS作業正常運行。否則屬於不相容的情況，CTAS作業會拋出異常。具體規則如下：

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

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

• FLOAT和DOUBLE歸一化為DOUBLE。

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

程式碼範例：

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

將MongoDB源表同步到Hologres表

使用限制：

• Realtime ComputeFlink VVR版本要求8.0.6及以上，MongoDB資料庫版本為6.0及以上。

• 在SQL Hints中需設定參數scan.incremental.snapshot.enabled和scan.full-changelog為true。

• MongoDB資料庫需要開啟前像後像（Pre- and Post-images）記錄功能，開啟方法參見Document Preimages。

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

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

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

程式碼範例：

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;