Realtime Compute for Apache Flink:CREATE DATABASE AS statement

Example 1: Synchronization of data from all tables in a database

In most cases, the CREATE DATABASE AS statement is used together with the catalog of the data source and the catalog of the destination store. For example, you can execute the CREATE DATABASE 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 tables 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. A MySQL database named tpcds exists, and the tpcds database has 24 tables. You can execute the following statement to synchronize all tables from the tpcds database to Hologres. The synchronized data includes future data changes and table schema changes. You do not need to create tables in Hologres in advance.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds  -- Create a database named holo_tpcds in Hologres. 
WITH ('sink.parallelism' = '4') -- Configure the parameters of the destination database. By default, the parallelism for the Hologres sink is set to 4. This setting is optional. 
AS DATABASE mysql.tpcds INCLUDING ALL TABLES -- Synchronize all tables from the tpcds database in the MySQL database. 
/*+ OPTIONS('server-id'='8001-8004') */ ; -- Configure additional parameters for MySQL CDC source tables. This setting is optional.

Example 2: Merging and synchronization of data of tables in a sharded database

If you want to merge and synchronize data of tables in a sharded database, you must use the database names that are defined in regular expressions to match the database shards from which data is synchronized. You can use the CREATE DATABASE AS statement to merge and synchronize data from the tables that have the same name in multiple upstream database shards to the table that has the same name as the source table in the Hologres destination database. The database names and table names are written as two additional fields to each sink table. To ensure that the primary key is unique, the database name, table name, and original primary key are used together to become the new joint primary key of the Hologres table.

For example, if a MySQL instance contains multiple database shards from order_db01 to order_db99. Each database shard contains multiple tables, such as order and order_detail. You can use the following statement to synchronize data from all tables such as order and order_detail of the 99 database shards to Hologres. The data includes future data changes and table schema changes. You do not need to create a table in Hologres in advance.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order -- Create a database named holo_order in Hologres. The database contains all tables in the order_db database shards of the MySQL instance. 
WITH('sink.parallelism'='4')        -- Specify the parameters of the destination database. By default, the parallelism of each Hologres sink is 4. The setting is optional. 
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES -- Synchronize data from all tables in the order_db database shards of the MySQL instance. 
/*+OPTIONS('server-id'='8001-8004')*/;  -- Configure additional parameters for MySQL CDC source tables. This setting is optional.

Example 3: Synchronization of data from new tables in the source database

In VVR 8.0.6 or later, after a deployment that uses the CREATE DATABASE AS statement is started and new tables are added to the source database, you can restart the deployment from savepoints to capture the new tables. This way, the data in the new tables can be synchronized.

1. When you develop an SQL draft, add the following statement to enable the feature of reading data from new tables by using the CREATE DATABASE AS statement:

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

2. If data from a new table needs to be synchronized in a deployment, cancel the deployment and select Stop With Savepoint in the Cancel this deployment dialog box.

3. On the SQL Editor page, redeploy an SQL draft for the 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 4: Execution of multiple CREATE TABLE AS and CREATE DATABASE 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, if the MySQL instance has multiple databases named tpcds, tpch, and user_db01 to user_db99 (database shards), you can combine multiple CREATE TABLE AS and CREATE DATABASE AS statements to synchronize data from all databases and tables of the MySQL instance to Hologres. This way, you can run only one Flink deployment to synchronize data from all tables in the databases and use only one source operator to read data from all tables. The following code shows an example.

USE CATALOG holo;

BEGIN STATEMENT SET;

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

-- Synchronize data from the tpcds database. 
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;
-- Synchronize data from the tpch database. 
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */ ;

END;

Example 5: Execution of multiple CREATE DATABASE AS statements to synchronize data from all tables in a database to Kafka

When you execute multiple CREATE DATABASE AS statements to synchronize data from all tables in a database to Kafka, you must configure the cdas.topic.pattern parameter to prevent topic conflicts because the same table may exist in different databases. The cdas.topic.pattern parameter specifies the format of the name of a topic. You can replace the placeholder {table-name} with a table name. For example, if you configure 'cdas.topic.pattern'='db1-{table-name}' and the name of the upstream table is named table1, the name of the topic in Kafka that corresponds to the table1 table is db1-table1.

For example, a MySQL instance has multiple databases, such as tpcds and tpch. You can execute the following statements to synchronize data from all databases and tables in the MySQL instance to Kafka and prevent topic conflicts during data synchronization. Sample code:

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- Synchronize data from the tpcds database. 
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') */ ;

-- Synchronize data from the tpch database. 
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;

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.