E-MapReduce:Use the Flink service in a Dataflow cluster to synchronize data from MySQL to StarRocks by using the CTAS statement

Supported schema changes

Unsupported schema changes

• Data type changes (for example, VARCHAR to BIGINT, or NOT NULL to NULLABLE)

• Constraint changes (primary key or index)

• Addition or deletion of a non-nullable column

If any unsupported schema change occurs in MySQL, delete the destination table in StarRocks and restart the CTAS job. StarRocks will recreate the table and resync all historical data.

Because CTAS uses record-level schema comparison, deleting then re-adding a column with no intervening data changes is treated as no schema change. The schema sync triggers only when new data arrives.

For supported field type mappings between Flink and StarRocks, see Continuously load data from Apache Flink®.

Start a YARN session and open the SQL Client

1. SSH into the Dataflow cluster. For more information, see Log on to a cluster.

2. Go to the Flink directory and start a YARN session in detached mode:

   cd /opt/apps/FLINK/flink-current
   ./bin/yarn-session.sh --detached

   A successful start prints application_XXXX_YY in the output. This is the session ID.

   

3. Open the SQL Client with the session ID:

   ./bin/sql-client.sh -s <application_XXXX_YY>

   Replace <application_XXXX_YY> with the session ID from the previous step.

Create catalogs for MySQL and StarRocks

Run the following statements in the SQL Client to register both data sources as catalogs:

CREATE CATALOG sr WITH (
  'type'     = 'starrocks',
  'endpoint' = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
  'username' = 'test',
  'password' = '1qaz!QAZ',
  'dbname'   = 'test_cdc'
);

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

StarRocks catalog parameters

MySQL catalog parameters

Choose a delivery semantics and run the CTAS statement

Choose one of the following three modes based on your requirements, then run the corresponding CTAS statement in the SQL Client.

USE CATALOG sr;

CREATE TABLE IF NOT EXISTS runoob_tbl1 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://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
  'load-url'                          = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
  'table-name'                        = 'runoob_tbl_sr',
  'username'                          = 'test',
  'password'                          = '1qaz!QAZ',
  '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'
) */;

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_tbl1 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://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
  'load-url'                          = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
  'table-name'                        = 'runoob_tbl',
  'username'                          = 'test',
  'password'                          = '1qaz!QAZ',
  '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'
) */;

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://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
  'load-url'                          = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
  'table-name'                        = 'runoob_tbl_sr',
  'username'                          = 'test',
  'password'                          = '1qaz!QAZ',
  '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'
) */;

WITH clause parameters

If you use a Flink version earlier than vvr-6.0.5-flink-1.15, add 'sink.use.new-apiapi' = 'false' to the WITH clause. For other configuration options, see Continuously load data from Apache Flink.

OPTIONS clause parameters

All examples use the OPTIONS hint to pass MySQL CDC connector settings inline with the CTAS statement.

Verify initial data

Connect to StarRocks and query the destination table:

USE test_cdc;
SELECT * FROM runoob_tbl1;

Expected output:

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

Verify inserts

Insert a row in MySQL:

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

Query StarRocks to confirm the row was replicated:

SELECT * FROM runoob_tbl1;

Expected output:

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

Verify updates

Update a row in MySQL:

UPDATE runoob_tbl SET runoob_title = 'new' WHERE runoob_id = 18;

Query StarRocks to confirm the update was replicated:

SELECT * FROM runoob_tbl1;

Expected output:

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

Verify deletes

Delete a row in MySQL:

DELETE FROM runoob_tbl WHERE runoob_id = 1;

Query StarRocks to confirm the deletion was replicated:

SELECT * FROM runoob_tbl1;

Expected output:

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

Verify schema changes

Add a nullable column in MySQL:

ALTER TABLE `runoob_tbl` ADD COLUMN `add_col2` INT;

Insert a row that populates the new column:

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

Query StarRocks to confirm the schema change was propagated:

SELECT * FROM runoob_tbl1;

Expected output:

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

The new add_col2 column appears in StarRocks. The existing row (runoob_id=18) shows NULL because the column was not present when that row was last written.