E-MapReduce:Use CTAS and CDAS in Realtime Compute for Flink to synchronize data from MySQL to StarRocks

Schema change behavior

CTAS uses a fixed policy for propagating schema changes. The following tables summarize what is and is not propagated automatically.

Supported schema changes

Unsupported schema changes

If any of the following changes occur in the source table, delete the destination table in StarRocks and restart the CTAS job. Flink recreates the table and resynchronizes all historical data.

• Change a column's data type (for example, VARCHAR to BIGINT, or NOT NULL to NULLABLE)

• Change constraints such as the primary key or index

• Add or delete a non-nullable column

• Adjust field lengths in DDL (Data Definition Language) statements

CTAS detects schema changes by comparing the schema of consecutive data records — it does not parse DDL statement types. As a result:

If a column is deleted and then re-added with no data changes in between, CTAS treats this as no schema change.

A schema change is only propagated when new data arrives in the source table after the change.

For field type mappings, see Continuously load data from Apache Flink.

When CTAS merges multiple MySQL tables, Flink automatically prepends _db_name and _table_name columns to the destination table to track the source. Define your own column order starting from the third column.

At-least-once semantics (recommended for low-latency scenarios)

Data is written at a configurable flush interval. Memory usage is lower, but duplicate writes may occur on failure.

/* 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 (recommended for data-critical scenarios)

No data loss or duplication on failure. Data visibility depends on the checkpoint interval.

/* 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'
) */;

For checkpoint configuration options, see Checkpointing.

Simple mode (recommended for quick setup)

Flink infers the table definition from the MySQL schema — no need to specify engine, key, or distribution manually. Partitioned tables are not supported in simple mode; create partitions using normal mode instead.

/* 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'      = 'test',
  'password'      = '123456',
  'database-name' = 'test_cdc',
  'table-name'    = 'runoob_tbl'
) */;

WITH clause parameters

If your Flink version is earlier than vvr-6.0.5-flink-1.15, add 'sink.use.new-apiapi' = 'false' to the WITH clause. For additional sink parameters, see Continuously load data from Apache Flink.

OPTIONS clause parameters

The OPTIONS clause configures the MySQL CDC source.

Publish and start the deployment

1. On the Advanced tab of the Draft Editor page, set Engine Version to vvr-6.0.3-flink-1.15 or later.

2. In the upper-right corner of the Draft Editor page, click Publish.

3. On the Deployments page, find the new deployment and click Start in the Actions column.

Verify initial data

Query the StarRocks table to confirm the seed row was synchronized:

SELECT * FROM runoob_tbl1;

Expected output:

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

Verify INSERT

On the SQL Console tab of the ApsaraDB RDS for MySQL instance, insert a row:

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 both rows appear:

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 UPDATE

On the MySQL SQL Console, update a row:

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

Query StarRocks to confirm the change:

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 DELETE

On the MySQL SQL Console, delete a row:

DELETE FROM runoob_tbl WHERE runoob_id = 1;

Query StarRocks to confirm the row is removed:

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 change: add a nullable column

On the MySQL SQL Console, add a nullable column and insert a row with a value in that column:

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

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 new column appears and existing rows show NULL:

SELECT * FROM runoob_tbl1;

Expected output:

+-----------+--------------+---------------+-----------------+---------+----------+
| 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 |
+-----------+--------------+---------------+-----------------+---------+----------+

The add_col2 column was automatically added to StarRocks when the first data row containing it arrived.