CREATE DATABASE AS（CDAS）语句

背景信息

使用限制

• 仅Flink计算引擎vvr-4.0.11-flink-1.13及以上版本支持CDAS语法。

  重要

  CDAS语法不支持进行调试。

• 仅Flink计算引擎vvr-4.0.13-flink-1.13及以上版本支持分库合并同步。

• CDAS语法不支持MiniBatch配置。

  对于已部署的SQL作业，若需要调整MiniBatch配置，请重新创建新的SQL作业，而非直接修改现有作业的配置并重启。

  重要

  在创建SQL作业之前，请确保配置管理页面的作业默认配置页签中的其他配置处删除了MiniBatch配置。

• CDAS支持的上下游存储列表如下。

  连接器名称	源表	结果表	备注

  连接器名称	源表	结果表	备注
  MySQL	√	×	不支持同步MySQL视图。
  消息队列Kafka	√	×	无。
  MongoDB	√	×	暂不支持分库合并同步。 暂不支持同步MongoDB元信息。 支持通过CDAS语句将MongoDB中的数据及表结构变更同步至目标表。具体的配置要求请参见使用MongoDB Catalog。
  Upsert Kafka	×	√	无。
  实时数仓Hologres	×	√	如果下游是Hologres，CDAS在默认情况下会为每个表创建相应数量（connectionSize参数值）个连接。此时您可以使用connectionPoolName参数，让配置相同名称连接池的表可以共享连接池。 说明 在将数据同步到Hologres时，如果您的上游源表包含了Fixed Plan不支持类型的数据，建议通过INSERT INTO语句的方式，在Flink内部做类型转换后将数据同步到Hologres。不要用CDAS方式创建Sink结果表进行数据同步，因为这种方式会无法走Fixed Plan，写入性能较差。 实时计算Flink版仅支持读写Hologres内表，因此Hologres实例必须是独占实例，不支持Hologres共享集群实例。
  StarRocks	×	√	仅支持EMR的StarRocks。
  流式数据湖仓Paimon	×	√	仅实时计算引擎VVR 6.0.7及以上版本支持Paimon结果表。 仅实时计算引擎VVR 8.0.10及以上版本支持同步到Paimon DLF 2.0结果表。

前提条件

• 执行CDAS语法前，确保工作空间中已注册目标端的Catalog，详情请参见数据管理。

• 执行CDAS语法前，如果您需要访问不同账号下的上下游资源、以及使用RAM用户或RAM角色等身份访问时，请确保登录Flink全托管的账号具有读写上下游资源的权限，否则会因为权限不足导致读写操作失败。

注意事项

• 使用VVR 8.0.6及以上版本时，CDAS作业启动后，支持添加新表后从作业快照重启，从而捕获到新的表。详情请参见示例三：源库新增表加入数据同步。

• 使用VVR 8.0.5及以下版本时，CDAS作业启动后，作业同步的表已经确定，数据库中新增的表不会自动捕捉，也无法通过重启作业的方式捕获到。如果需要同步新增的表，您可以选择以下任一种方案：

  • 原有CDAS作业不变，启动一个新的作业同步新增的表。例如

    // 新建CTAS作业同步新增加的表new_table
    CREATE TABLE IF NOT EXISTS new_table
    AS TABLE mysql.tpcds.new_table 
    /*+ OPTIONS('server-id'='8008-8010') */;

  • 停止现有CDAS作业，清理已同步的数据后，以全新状态重启CDAS作业来重新同步数据。

功能特性

启动流程

当执行CDAS语句时，阿里云Flink将会按照以下流程执行：

1. 检查目标存储中是否存在目标库和结果表。

   • 如果不存在目标库，则通过目标端Catalog去目标存储中创建相应的目标库。

   • 如果存在目标库，则跳过建库，并检查目标库是否存在该结果表。

     • 如果不存在，则在目标库中创建相应的结果表，该结果表具有和源库中表相同的表名和Schema。

     • 如果存在，则跳过建表。

2. 提交和启动相应的数据同步作业。将源库中的数据以及Schema变更同步到目标库下的表中。

例如，从MySQL到Hologres的CDAS数据同步流程如下图所示。

表结构变更同步策略

因为CDAS是CTAS语法的一个语法糖，所以表结构变更能力与CTAS一致，详情请参见CREATE TABLE AS（CTAS）语句。

基本语法

CREATE DATABASE IF NOT EXISTS <target_database>
[COMMENT database_comment]
[WITH (key1=val1, key2=val2, ...)]
AS DATABASE <source_database>
INCLUDING { ALL TABLES | TABLE 'table_name' }
[EXCLUDING TABLE 'table_name']
[/*+ OPTIONS(key1=val1, key2=val2, ... ) */]

<target_database>:
  [catalog_name.]db_name

<source_database>:
  [catalog_name.]db_name

CDAS语法复用了CREATE DATABASE语法的基本结构，其中的参数解释如下表所示。

示例

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源表的额外参数。

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源表的额外参数。

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;

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;

相关文档

• 为什么主键为bigint unsigned的MySQL表注册Flink Catalog，主键会变为decimal？但是使用CTAS同步到Hologres后，主键又变为text？

• 如果您需要进行单表同步、分库分表合并同步或自定义计算列同步，请参见CREATE TABLE AS（CTAS）语句。

• 使用CTAS和CDAS实现数据同步的教程详情请参见数据库实时入仓快速入门。