Realtime Compute for Apache Flink:CREATE TABLE AS statement

Example 1: single-table synchronization

In most cases, the CREATE TABLE AS statement is used with the catalog of the data source and the catalog of the destination store. For example, you can execute the CREATE TABLE AS statement and use a MySQL catalog and a Hologres catalog to synchronize full data and incremental data from the MySQL database to Hologres. You can use a MySQL catalog to parse the schema and related parameters of the source table without the need to manually write DDL statements.

For example, a Hologres catalog named holo and a MySQL catalog named mysql are created in your workspace. You can use the following code to synchronize data from the web_sales table of the MySQL database to Hologres:

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales  
WITH ('jdbcWriteBatchSize' = '1024')   -- Configure the parameters of the sink table. This setting is optional. 
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */; -- Configure additional parameters for the MySQL CDC source table.

Example 2: merging and synchronization of multiple tables in a sharded database

If you want to merge and synchronize data of multiple tables in a sharded database, you can use a MySQL catalog and a regular expression that defines database shard names and table names to match the tables whose data you want to synchronize. You can use the CREATE TABLE AS statement to merge multiple tables in a sharded database into a Hologres table. The database shard names and table names are written as the values of two additional fields to the table. To ensure that the primary key is unique, the database shard name field, table name field, and original primary key are used as the new joint primary key of the Hologres table.

USE CATALOG holo;

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

The following figure shows the effect of the merging operation. If you add a column named age to the table named user02 and insert a data record into the table, the data and schema changes on the user02 table can be automatically synchronized to the sink table in real time even if the schemas of the source tables are different.

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

Example 3: addition of custom computed columns

This example describes how to add computed columns to the source tables for data conversion and computation during the merging and synchronization of multiple tables in a sharded database. In this example, the user table is used.

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

The following figure shows the synchronization effect.

Example 4: execution of multiple CREATE TABLE AS statements

Realtime Compute for Apache Flink allows you to use the STATEMENT SET statement to commit multiple CREATE TABLE AS statements as one deployment. Realtime Compute for Apache Flink can also optimize the data of source operators and use a single source operator to read data from multiple business tables. This operation can help reduce the use of server-id, the number of database connections, and the database reading load. Therefore, this operation is suitable for MySQL CDC data sources.

For example, data is synchronized from the web_sales table in the first code segment, and data is synchronized from multiple tables whose names start with user in a sharded database in the second code segment. In this case, you can use the STATEMENT SET statement to commit these code segments as one deployment.

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Synchronize data from the web_sales table. 
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

-- Synchronize data from multiple tables in the specified database shards whose names start with user. 
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Example 5: execution of multiple CREATE TABLE AS statements to synchronize data from the same source table to different sink tables

In Realtime Compute for Apache Flink that uses VVR 4.0.16 or later, you can execute multiple CREATE TABLE AS statements to synchronize data from the same data source table to different sink tables without the need to add computed columns.

USE CATALOG `holo`;

BEGIN STATEMENT SET;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user table in database1 of Hologres.
CREATE TABLE IF NOT EXISTS `database1`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user table in database2 of Hologres.
CREATE TABLE IF NOT EXISTS `database2`.`user`
AS TABLE `mysql`.`tpcds`.`user`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

If you want to add a computed column to the sink table, execute the following statements to perform data synchronization:

-- Create a temporary table named user_with_changed_id based on the source table user. You can define a computed column. The following sample code defines the computed_id column that is calculated based on the id column of the source table. 
CREATE TEMPORARY TABLE `user_with_changed_id` (
  `computed_id` AS `id` + 1000
) LIKE `mysql`.`tpcds`.`user`;

-- Create a temporary table named user_with_changed_age based on the source table user. You can define a computed column. The following sample code defines the computed_age column that is calculated based on the age column of the source table. 
CREATE TEMPORARY TABLE `user_with_changed_age` (
  `computed_age` AS `age` + 1
) LIKE `mysql`.`tpcds`.`user`;

BEGIN STATEMENT SET;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user_with_changed_id table of Hologres. The user_with_changed_id table contains the IDs that are obtained from the calculation based on the id column of the source table. The obtained IDs are in the computed_id column.  
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
AS TABLE `user_with_changed_id`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Execute the CREATE TABLE AS statement to synchronize data from the user table of the MySQL database to the user_with_changed_age table of Hologres. The user_with_changed_age table contains the age values that are obtained from the calculation based on the age column of the source table. The obtained age values are in the computed_age column.  
CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
AS TABLE `user_with_changed_age`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Example 6: addition of a CREATE TABLE AS statement for a data synchronization deployment that executes multiple CREATE TABLE AS statements

In Realtime Compute for Apache Flink that uses VVR 8.0.1 or later, if you add a CREATE TABLE AS statement for a deployment that executes multiple CREATE TABLE AS statements after the deployment is started, you can restart the deployment from a savepoint to capture the new table and synchronize data from the new table.

1. When you develop an SQL draft, you must add the following statement to enable the feature of reading data from a new table:

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

2. If you want to add a CREATE TABLE AS statement for a deployment, find the deployment on the Deployments page and click Cancel in the Actions column. In the message that appears, select Stop With Savepoint and click OK.

3. On the SQL Editor page, add the CREATE TABLE AS statement to the SQL editor of the deployment, and click Deploy in the upper-left corner to redeploy the SQL deployment.

4. On the Deployments page, click the name of the deployment. On the deployment details page, click the Status tab. Then, click the History tab.

5. In the Savepoints list, find the savepoint that was created when the deployment was canceled.

6. Choose More > Start job from this snapshot in the Actions column.

7. In the Start Job panel, configure the parameters. For more information, see Start a deployment.

Example 7: execution of the CREATE TABLE AS statement to synchronize data from a source table to a Hologres partitioned table

When you create a Hologres partitioned table that has a primary key, the partition field must be a field in the primary key. If you want to synchronize data from a MySQL table to Hologres, execute the following statement to create the table:

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

When you execute the CREATE TABLE AS statement to synchronize data from the source table to the Hologres partitioned table, the processing mechanism varies based on whether the primary key of the source table contains a partition field of the sink table:

• If the primary key of the source table contains a partition field of the sink table, you can execute the following SQL statement for data synchronization. In this example, the partition field in the Hologres partitioned table is product_id.

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

• If the primary key of the source table does not contain a partition field of the sink table, the primary key of the MySQL table is used when the Hologres table is created. In this case, an error occurs. In this example, the partition field in the Hologres partitioned table is city. In this case, you can declare the primary key in the CREATE TABLE AS statement and specify the primary key of the Hologres partitioned table to make the deployment run as expected. Sample statement:

  -- Execute the following SQL statement to declare the order_id, product_id, and city fields as the primary key of the Hologres partitioned table: 
  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`;

Example 8: execution of the CREATE TABLE AS statement to synchronize data to a Hologres table in type normalization mode

When you execute the CREATE TABLE AS statement, you may need to change the data type precision for the existing fields. For example, you can change the data type precision from VARCHAR(10) to VARCHAR(20).

• For Realtime Compute for Apache Flink whose engine version is earlier than vvr-6.0.5-flink-1.15, if you change the data type of a field in the source table of a deployment in which the CREATE TABLE AS statement is executed, the deployment may fail. In this case, you must recreate the sink table.

• When you synchronize data in a deployment of Realtime Compute for Apache Flink whose engine version is vvr-6.0.5-flink-1.15 or later, the type normalization mode is supported. We recommend that you enable the type normalization mode the first time you start a deployment that executes the CREATE TABLE AS statement. If you do not enable the type normalization mode the first time you start the deployment, the type normalization mode does not take effect. In this case, you must delete the downstream table and restart the deployment without using the state data.

  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders` 
  WITH (
  'connector' = 'hologres', 
  'enableTypeNormalization' = 'true' -- Use the type normalization mode. 
  ) AS TABLE `mysql`.`tpcds`.`orders`;

  In type normalization mode, data type changes in the source table do not lead to a deployment failure only if the data types before and after the change can be converted into the same data type based on type normalization rules. If you use the type normalization mode, take note of the following items:

  • The TINYINT, SMALLINT, INT, and BIGINT data types are converted into the BIGINT data type.

  • The CHAR, VARCHAR, and STRING data types are converted into the STRING data type.

  • The FLOAT and DOUBLE data types are converted into the DOUBLE data type.

  • Other data types are converted based on the data type mappings between Hologres and Flink fields. For more information, see Data type mappings.

  Example:

  • When the type normalization mode is enabled, the SMALLINT and INT data types are converted into the BIGINT data type. If you change the SMALLINT data type to the INT data type, the change is considered successful. Therefore, the deployment that executes the CREATE TABLE AS statement runs as expected.

  • When the type normalization mode is enabled, the FLOAT data type is converted into the DOUBLE data type and the BIGINT data type is converted into the BIGINT data type. If you change the FLOAT data type to the BIGINT data type, a data type incompatibility error occurs.

Example 9: synchronization of data from a MongoDB source table to a Hologres table by using the CREATE TABLE AS statement

In Realtime Compute for Apache Flink that uses VVR 8.0.6 or later, you can execute the CREATE TABLE AS statement to synchronize data from a MongoDB source table to a destination table in real time and synchronize the schema changes in the source table to the destination table. You can use MongoDB catalogs together with the CREATE TABLE AS statement to synchronize schema changes. You do not need to manually define the table schema. For more information about MongoDB catalogs, see Manage MongoDB catalogs.

The following sample code shows how to use the CREATE TABLE AS statement to synchronize data from a MongoDB source table to a Hologres table.

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;

If you want to synchronize data from multiple MongoDB collections in a deployment, make sure that the configurations of the following parameters for all tables in the deployment are the same:

• Parameters related to the MongoDB database, including hosts, scheme, username, password, and connectionOptions

• scan.startup.mode

Example 10: synchronization of data from all tables in a MySQL database to Kafka

Multiple deployments may use the same MySQL table. If multiple deployments use the same MySQL table, the MySQL database establishes multiple connections. This causes a heavy load on the MySQL server and network. To reduce the load on an upstream MySQL database, Realtime Compute for Apache Flink allows you to synchronize data from all tables in the MySQL database to Kafka. This way, Kafka is used as the intermediate layer for data synchronization. You can execute the CREATE DATABASE AS statement to synchronize data from all tables in a MySQL database to Kafka. You can also execute the CREATE TABLE AS statement to synchronize data from a single table in a MySQL database to Kafka. For more information, see Synchronize data from all tables in a MySQL database to Kafka.