All Products
Search

This Product

    Realtime Compute for Apache Flink:Flink CDC data ingestion job

    Document Center

    Realtime Compute for Apache Flink:Flink CDC data ingestion job

    Last Updated:Jun 21, 2026

    Realtime Compute for Apache Flink provides a powerful data ingestion feature based on Flink CDC. This guide demonstrates how to build a Flink CDC data ingestion job to synchronize an entire MySQL database to a StarRocks database.

    Prerequisites

    Background information

    Assume that your ApsaraDB RDS for MySQL instance has a database named order_dw_mysql that contains three business tables: orders, orders_pay, and product_catalog. To synchronize these tables and their data to the order_dw_sr database in StarRocks, follow these steps:

    1. Step 1: Prepare test data in ApsaraDB RDS for MySQL

    2. Step 2: Develop a Flink CDC data ingestion job

    3. Step 3: Start the Flink CDC data ingestion job

    4. Step 4: Verify the synchronization results in StarRocks

    Step 1: Prepare MySQL test data

    1. Create a database and an account.

      Create a database named order_dw_mysql and a standard account with read and write permissions to it. For more information, see (Deprecated, redirected to "Step 1") Create a database and an account and Manage databases.

    2. Log on to the ApsaraDB RDS for MySQL instance by using Data Management (DMS).

    3. In the SQL Console window, enter the following commands and click Execute to create three business tables and insert data.

      CREATE TABLE `orders` (
  order_id bigint not null primary key,
  user_id varchar(50) not null,
  shop_id bigint not null,
  product_id bigint not null,
  buy_fee numeric(20,2) not null,   
  create_time timestamp not null,
  update_time timestamp not null default now(),
  state int not null 
);
CREATE TABLE `orders_pay` (
  pay_id bigint not null primary key,
  order_id bigint not null,
  pay_platform int not null, 
  create_time timestamp not null
);
CREATE TABLE `product_catalog` (
  product_id bigint not null primary key,
  catalog_name varchar(50) not null
);
-- Prepare data
INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
INSERT INTO orders VALUES
(100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
(100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
(100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
(100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
(100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
(100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
(100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
INSERT INTO orders_pay VALUES
(2001, 100001, 1, '2023-02-15 17:40:56'),
(2002, 100002, 1, '2023-02-15 17:40:56'),
(2003, 100003, 0, '2023-02-15 17:40:56'),
(2004, 100004, 0, '2023-02-15 17:40:56'),
(2005, 100005, 0, '2023-02-15 18:40:56'),
(2006, 100006, 0, '2023-02-15 18:40:56'),
(2007, 100007, 0, '2023-02-15 18:40:56');

    Step 2: Develop a Flink CDC job

    1. Log on to the Realtime Compute for Apache Flink management console.

    2. Click Console to go to the project workspace.

    3. In the left-side navigation pane, choose Development > Data Ingestion.

    4. Click the image icon, click New Draft with Template, select MySQL to StarRocks data synchronization, and then click Next.

    5. Enter a Job Name and Storage Location, select an Engine Version, and then click OK.

    6. Configure the YAML job code.

      The following code provides an example to synchronize all tables from the order_dw_mysql database in MySQL to the order_dw_sr database in StarRocks.

      source:
  type: mysql
  hostname: rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: order_dw_mysql.\.*
  server-id: 8601-8604
  # (Optional) Synchronize data from tables that are newly created during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate reading.
  scan.only.deserialize.captured.tables.changelog.enabled: true 
sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030
  load-url: fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030
  username: ${secret_values.starrocksusername}
  password: ${secret_values.starrockspassword}
  table.create.properties.replication_num: 1
  sink.buffer-flush.interval-ms: 5000 # Flush data every 5 seconds.
route:
  - source-table: order_dw_mysql.\.*
    sink-table: order_dw_sr.<>
    replace-symbol: <>
    description: route all tables in source_db to sink_db
pipeline:
  name: Sync MySQL Database to StarRocks

      The following table describes the configuration parameters required for this example. For more information about data ingestion parameters, see MySQL and StarRocks.

      Note

      YAML jobs only support project variables. You can use variables to prevent information such as passwords from being displayed in plaintext. For more information, see Variable Management.

      Category

      Parameter

      Description

      Example value

      source

      hostname

      The IP address or hostname of the MySQL database.

      We recommend using the internal endpoint.

      rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com

      port

      The port number of the MySQL database service.

      3306

      username

      The username and password for the MySQL database. Use the account credentials created in Step 1: Prepare test data in ApsaraDB RDS for MySQL.

      ${secret_values.mysqlusername}

      password

      ${secret_values.mysqlpassword}

      tables

      The names of the MySQL tables. You can use regular expressions to read data from multiple tables.

      In this topic, all tables and data in the order_dw_mysql database are synchronized.

      order_dw_mysql.\.*

      server-id

      A unique numeric ID for the database client connection.

      8601-8604

      sink

      jdbc-url

      The JDBC connection URL.

      Specify the IP address and query port of the Frontend (FE) in the format jdbc:mysql://ip:port.

      On the Instance Details tab in the E-MapReduce console, you can view the internal endpoint and query port of the FE for the target instance.

      jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030

      load-url

      The HTTP service URL used to connect to the FE node.

      On the Instance Details tab in the E-MapReduce console, you can view the internal endpoint and HTTP port of the FE for the target instance.

      fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030

      username

      The credentials for connecting to StarRocks.

      Use the credentials that you configured when you created the StarRocks instance.

      Note

      This example uses variables to avoid exposing credentials in plaintext. For more information, see Manage variables.

      ${secret_values.starrocksusername}

      password

      ${secret_values.starrockspassword}

      sink.buffer-flush.interval-ms

      The flush interval for the internal buffer.

      A short interval (5 seconds) is used because this example contains a small amount of data, allowing you to see the results quickly.

      5000

      route

      source-table

      The source table or tables to route.

      You can use a regular expression to match multiple tables. For example, order_dw_mysql.\.* routes all tables in the order_dw_mysql database.

      order_dw_mysql.\.*

      sink-table

      The destination table pattern for the routed data.

      You can use the symbol defined in the replace-symbol parameter as a placeholder for each source table name to achieve many-to-many routing.

      For more information about routing rules, see Route module.

      order_dw_sr.<>

      replace-symbol

      The placeholder for the source table name used in pattern matching.

      <>

    7. Click Deploy.

    Step 3: Start the Flink CDC job

    1. On the Data Ingestion page, click Deploy and then click OK in the dialog box that appears.

    2. On the O&M > Deployments page, find the target YAML job and click Start in the Actions column.

    3. Click Start.

      In this example, select Initial Mode. For more information about the parameters, see Start a job. After the job starts, you can monitor its status on the Deployments page.

    Step 4: Verify results in StarRocks

    After the job enters the RUNNING state, you can verify the data in StarRocks.

    1. Connect to an EMR Serverless StarRocks instance by using EMR StarRocks Manager.

    2. In the left-side navigation pane, click SQL Editor. On the Database tab, click the image refresh icon.

      A database named order_dw_sr appears under default_catalog.

    3. On the Query List tab, click + File to create a Query Script. Enter the following SQL statements and click Run.

      SELECT * FROM default_catalog.order_dw_sr.orders order by order_id;
SELECT * FROM default_catalog.order_dw_sr.orders_pay order by pay_id;
SELECT * FROM default_catalog.order_dw_sr.product_catalog order by product_id;

    4. View the results below the commands.

      The results show that the tables and data from the MySQL database now exist in StarRocks.

      The synchronized tables include default_catalog.order_dw_sr.orders, default_catalog.order_dw_sr.orders_pay, and default_catalog.order_dw_sr.product_catalog. You can run SELECT statements to query each table and verify data integrity.

    Related documentation