实时计算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，利用正则表达式的表名和库名来匹配所要同步的多张表。使用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语句作为一个作业提交

实时计算Flink版支持使用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表

实时计算Flink 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数据库的压力，实时计算Flink版提供MySQL整库同步到Kafka的能力，通过引入Kafka作为中间层，并使用CDAS整库同步或CTAS整表同步到Kafka来解决。具体操作请参见MySQL整库同步Kafka。