E-MapReduce:Synchronize data from MySQL to StarRocks by using CTAS and CDAS

Prerequisites

• You have activated Realtime Compute for Apache Flink (Fully Managed) and created a Flink cluster. For more information, see Activate Realtime Compute for Apache Flink (Fully Managed) and Getting started with a Flink SQL deployment.

• You have created a StarRocks cluster. For more information, see Create a StarRocks cluster.

• You have created an ApsaraDB RDS for MySQL instance. For more information, see Create an ApsaraDB RDS for MySQL instance.

Limitations

• The Flink cluster, StarRocks cluster, and ApsaraDB RDS for MySQL instance must be in the same VPC.

• The ApsaraDB RDS for MySQL instance must be version 5.7 or later.

• The StarRocks cluster must have internet access enabled.

• The Flink version in your Flink cluster must be 1.15-vvr-6.0.3 or later.

Step 1: Prepare test data

1. Create a test database and account. For more information, see Create databases and accounts for an ApsaraDB RDS for MySQL instance.

   After you create the database and account, grant read and write permissions to the test account.

   Note

   In this topic, the database is named test_cdc and the account is named test.

2. Use the test account to connect to the MySQL instance. For more information, see Use DMS to log on to an ApsaraDB RDS for MySQL instance.

3. Run the following commands in MySQL to create a data table.

   use test_cdc;
   
   CREATE TABLE IF NOT EXISTS `runoob_tbl`(
      `runoob_id` INT UNSIGNED AUTO_INCREMENT,
      `runoob_title` VARCHAR(100) NOT NULL,
      `runoob_author` VARCHAR(40) NOT NULL,
      `submission_date` DATE,
      `add_col` int DEFAULT NULL,
      PRIMARY KEY ( `runoob_id` )
   )ENGINE=InnoDB DEFAULT CHARSET=utf8;
   
   
   INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2022-06-22 17:13:44',3)

4. Log on to the StarRocks cluster by using SSH. For more information, see Log on to a cluster.

5. Run the following command to connect to the StarRocks cluster.

   mysql -h127.0.0.1 -P 9030 -uroot

6. Run the following commands to create a user and grant permissions.

   CREATE DATABASE test_cdc;
   CREATE USER 'test' IDENTIFIED by '123456';
   GRANT CREATE TABLE ON DATABASE test_cdc TO test;

Step 2: Create catalogs

• MySQL catalog

  • Example

    CREATE CATALOG mysql WITH (
      'type' = 'mysql',
      'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
      'port' = '3306',
      'username' = 'emr-test',
      'password' = '123456',
      'default-database' = 'test_cdc'
    );

  • Parameters

    Parameter	Description
    type	The catalog type. Set the value to mysql.
    hostname	The internal endpoint of the ApsaraDB RDS for MySQL instance. You can copy the internal endpoint from the Database Connection page of the instance in the ApsaraDB RDS console. For example, rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com.
    port	The port number of the MySQL database service. The default value is 3306.
    username	The username for accessing the MySQL database service. Use the username of the account created in Step 1: Prepare test data. This example uses test.
    password	The password for accessing the MySQL database service. Use the password of the account created in Step 1: Prepare test data.
    default-database	The name of the default MySQL database. Use the database name from Step 1: Prepare test data. This example uses test_cdc.

• StarRocks catalog

  • Example

    CREATE CATALOG sr  WITH (
      'type' = 'starrocks',
      'endpoint' = '172.16.**.**:9030',
      'username' = 'test',
      'password' = '123456',
      'dbname' = 'test_cdc'
    );

  • Parameters

    Parameter	Description
    type	The catalog type. Set the value to starrocks.
    endpoint	The IP address and port of the StarRocks Frontend (FE).
    username	The username for accessing the StarRocks cluster. Use the username of the account created in Step 1: Prepare test data. This example uses test.
    password	The password for the StarRocks database service. Use the password of the account created in Step 1: Prepare test data.
    dbname	The name of the StarRocks database. Use the database name from Step 1: Prepare test data. This example uses test_cdc.

Step 3: Create and publish a deployment

1. On the Draft Editor page of the Realtime Compute for Apache Flink console, write a CTAS statement.

   The following are three examples of CTAS statements.

   • At-least-once semantics: Use the sink.buffer-flush.interval-ms option to configure the interval at which data is written to StarRocks. This option reduces latency and memory usage.

     /*
           At-least-once semantics
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl_sr with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • Exactly-once semantics: You must configure a checkpoint interval. This prevents data loss and duplication during failures, but data visibility depends on the checkpoint interval. For more information, see Checkpointing.

     /*
           Exactly-once semantics.
     */
     set 'execution.checkpointing.interval' = '1 min';
     set 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
     set 'execution.checkpointing.timeout' = '10 min';
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl with (
     'starrocks.create.table.properties'=' engine = olap primary key(runoob_id) distributed by hash(runoob_id ) buckets 8',
     'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl',
     'username'='test',
     'password' = '123456',
     'sink.semantic' = 'exactly-once',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   • Simple mode: You do not need to define fields when you create the table. The table schema is copied from MySQL. However, you cannot create partitions. To use partitions, you must use normal mode.

     /*
           The two preceding examples use normal mode. This example demonstrates simple mode.
     */
     
     use CATALOG sr;
     
     CREATE TABLE IF NOT EXISTS runoob_tbl1 with (
     'starrocks.create.table.properties'='buckets 8',
     'starrocks.create.table.mode'='simple',
      'database-name'='test_cdc',
     'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
     'load-url'='172.16.**.**:18030',
     'table-name'='runoob_tbl_sr',
     'username'='test',
     'password' = '123456',
     'sink.buffer-flush.interval-ms' = '5000',
     'sink.properties.row_delimiter' = '\x02',
     'sink.properties.column_separator' = '\x01'
     )
      as table mysql.test_cdc.runoob_tbl  /*+ OPTIONS (   'connector' = 'mysql-cdc',
       'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'emr-test',
       'password' = '123456',
       'database-name' = 'test_cdc',
       'table-name' = 'runoob_tbl'  )*/;
                                     

   Table 1. WITH parameters

   Parameter	Required	Description
   starrocks.create.table.properties	Yes	The suffix definitions in the StarRocks CREATE TABLE statement, excluding the column definitions. Examples include engine, key, and buckets.
   database-name	Yes	The name of the StarRocks database. This example uses test_cdc.
   jdbc-url	Yes	Used to run query operations in StarRocks. For example, jdbc:mysql://172.16.**.**:9030. The172.16.**.** part is the internal IP address of the StarRocks cluster.
   load-url	Yes	The IP address and HTTP port of the StarRocks frontend (FE) in the format of Internal IP address of StarRocks cluster:Port. This topic uses port 8030 as an example. Select a port based on your cluster version: 18030: For EMR V5.9.0 or later and EMR V3.43.0 or later. 8030: For EMR V5.8.0 or earlier and EMR V3.42.0 or earlier. Note For more information about ports, see UI and ports.
   sink.semantic	No	Set this to exactly-once to guarantee data consistency semantics. The default is at-least-once.
   starrocks.create.table.mode	No	Supported values: normal (default): You must provide complete configurations, such as engine, key, and buckets, in the starrocks.create.table.properties option. simple: The engine is set to olap and the key type is set to primary key by default. The primary key is inherited from the MySQL table. By default, the table is distributed by a hash of all primary key columns, with no partitions. You must specify buckets in the starrocks.create.table.properties option. Other configurations like properties are optional.

   Note

   • Thesink.use.new-api parameter is removed in Flink versions 1.15-vvr-6.0.5 and later. If you use a version earlier than 1.15-vvr-6.0.5, you must add'sink.use.new-api'='false', to the WITH parameters.

   • For information about other configurations, see Continuously load data from Apache Flink.

   Table 2. OPTIONS parameters

   Parameter	Description
   connector	The connector type. Set the value to mysql-cdc.
   hostname	The internal endpoint of the ApsaraDB RDS for MySQL instance. You can copy the internal endpoint from the Database Connection page of the instance in the ApsaraDB RDS console. For example, rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com.
   port	The port number of the MySQL database service. The default value is 3306.
   username	The username for accessing the MySQL database service. Use the username of the account created in Step 1: Prepare test data. This example uses test.
   password	The password for accessing the MySQL database service. Use the password of the account created in Step 1: Prepare test data.
   table-name	The name of the table in StarRocks. Use the table name from Step 1: Prepare test data. This example uses runoob_tbl.
   database-name	The name of the default MySQL database. Use the database name from Step 1: Prepare test data. This example uses test_cdc.

2. In the Advanced settings on the Draft Editor page, select Flink version 1.15-vvr-6.0.3 or later.

3. Click online.

4. On the Deployments page, find the target deployment and click START in the Actions column.

Step 4: Scenario demonstration

Querying data

1. Log on to the StarRocks cluster by using SSH. For more information, see Log on to a cluster.

2. Run the following command to connect to the StarRocks cluster.

   mysql -h127.0.0.1 -P 9030 -uroot

3. In the StarRocks CLI, run the following commands to view the table data.

   use test_cdc;
   select * from runoob_tbl1;

   The output shows that data from the MySQL table has been synchronized to StarRocks.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Querying inserted data

1. In the SQL console for the ApsaraDB RDS for MySQL instance, run the following command to insert data.

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`)  values(1,'second','tom2','2022-06-23',1)

2. In the StarRocks CLI, run the following command to view the table data.

   select * from runoob_tbl1;

   The output shows that the data was inserted successfully.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | first        | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Synchronizing data updates

1. In the SQL console for the ApsaraDB RDS for MySQL instance, run the following command to update specific data.

   update runoob_tbl set runoob_title= 'new' where runoob_id = 18

2. In the StarRocks CLI, run the following command to view the table data.

   select * from runoob_tbl1;

   The output shows that the data update was synchronized.

   +-----------+--------------+---------------+-----------------+---------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col |
   +-----------+--------------+---------------+-----------------+---------+
   |         1 | second       | tom2          | 2022-06-23      |       1 |
   |        18 | new          | tom           | 2022-06-22      |       3 |
   +-----------+--------------+---------------+-----------------+---------+

Synchronizing data deletions

1. In the SQL console for the ApsaraDB RDS for MySQL instance, run the following command to delete specific data.

   DELETE FROM runoob_tbl WHERE runoob_id = 1

2. In the StarRocks CLI, run the following command to view the table data.

   select * from runoob_tbl1;

   The output shows that the data deletion was synchronized.

   +-----------+--------------+---------------+-----------------+---------+ 
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | 
   +-----------+--------------+---------------+-----------------+---------+
   |        18 | new          | tom           | 2022-06-22      |       3 | 
   +-----------+--------------+---------------+-----------------+---------+

Adding a nullable column

1. In the SQL console for the ApsaraDB RDS for MySQL instance, run the following command to add a nullable column.

   alter table `runoob_tbl` add COLUMN `add_col2` INT;

2. Run the following command to insert data.

   INSERT INTO runoob_tbl(`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`,`add_col2`)  values(1,'second','tom2','2022-06-23',1,2)

3. In the StarRocks CLI, run the following command to view the table data.

   select * from runoob_tbl1;

   The output shows that the schema change was successful.

   +-----------+--------------+---------------+-----------------+---------+----------+
   | runoob_id | runoob_title | runoob_author | submission_date | add_col | add_col2 | 
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |        18 | new          | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 
   |         1 | second       | tom2          | 2022-06-23      |       1 |      2   |
   |        18 | first        | tom           | 2022-06-22      |       3 |     NULL |
   +-----------+--------------+---------------+-----------------+---------+----------+ 

Introduction to CDAS

A CDAS statement is syntactic sugar for CTAS. It synchronizes an entire MySQL database to StarRocks as a single Flink deployment. You can also use the including table syntax to synchronize only a subset of tables.

CREATE DATABASE IF NOT EXISTS sr_db with (
  'starrocks.create.table.properties'=' buckets 8',
  'starrocks.create.table.mode'='simple',
  'jdbc-url'='jdbc:mysql://172.16.**.**:9030',
  'load-url'='172.16.**.**:18030',
  'username'='test',
  'password' = '123456',
  'sink.buffer-flush.interval-ms' = '5000' ,
  'sink.properties.row_delimiter' = '\x02',
  'sink.properties.column_separator' = '\x01'
)
as DATABASE mysql.test_cdc including table  'tabl1','tbl2','tbl3'
/*+ OPTIONS (   'connector' = 'mysql-cdc',
'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
'port' = '3306',   
'username' = 'test',
'password' = '123456',
'database-name' = 'test_cdc' )*/;