Realtime Compute for Apache Flink:Develop Flink CDC jobs (public preview)

Log on to the Realtime Compute for Apache Flink console.

Click Console in the Actions column of the target workspace.

In the left navigation pane, select Development > Data Ingestion.

Click , and then click Create From Template.

Select a data synchronization template. Currently supported templates: MySQL to StarRocks, MySQL to Paimon, and MySQL to Hologres.

Enter the job name, storage location, and engine version, and click OK.

Configure the source and sink parameters. See the documentation for the corresponding connector.

Log on to the Realtime Compute for Apache Flink console.

Click Console in the Actions column of the target workspace.

In the left navigation pane, select Data Studio > Data Ingestion.

Click , click New Draft From CTAS/CDAS, select the target CTAS or CDAS job, and click OK. The list shows only valid CTAS and CDAS jobs. Jobs with syntax errors and non-ETL jobs are excluded.

Enter the job name, storage location, and engine version, and click OK.

Log on to the Realtime Compute for Apache Flink console.

Click Console in the Actions column of the target workspace.

In the left navigation pane, select Development > Data Ingestion.

Click , select New Data Ingestion Draft, set the File Name and Engine Version, and click Create.

Copy the Flink CDC YAML from the open source community.

(Optional) Click Depth Check to validate syntax, network connectivity, and access permissions.

Log on to the Realtime Compute for Apache Flink console.

Click Console in the Actions column of the target workspace.

In the left navigation pane, select Data Development > Data Ingestion.

Click , select New Data Ingestion Draft, enter the File Name and Engine Version, and click Create.

Write the YAML configuration. A Flink CDC job consists of up to five blocks: The following template shows all five blocks with inline comments:

In a Flink CDC YAML job, separate each key and value with a space: key: value.

# Required
source:
  # Data source type
  type: <replace-with-source-connector-type>
  # Data source configurations — see the connector documentation for available parameters
  ...

# Required
sink:
  # Destination type
  type: <replace-with-sink-connector-type>
  # Destination configurations — see the connector documentation for available parameters
  ...

# Optional
transform:
  # Transformation rule for the flink_test.customers table
  - source-table: flink_test.customers
    # Projection: specify columns to include and apply computed expressions
    projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name
    # Filter: synchronize only rows where id > 10
    filter: id > 10
    description: append calculated columns based on source table

# Optional
route:
  # Map source table flink_test.customers to sink table db.customers_o
  - source-table: flink_test.customers
    sink-table: db.customers_o
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    description: sync customers_suffix table

# Optional
pipeline:
  name: MySQL to Hologres Pipeline

For the full syntax reference and all configuration parameters, see Development reference for Flink CDC data ingestion jobs. The following example synchronizes all tables from the app_db database in MySQL to Hologres. Credentials are stored as variables to avoid hardcoding sensitive information.

# Sync all tables in app_db from MySQL to Hologres
source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404
  # (Optional) Capture tables created after the job starts (incremental phase only)
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and column comments
  include-comments.enabled: true
  # (Optional) Dispatch unbounded chunks first to reduce TaskManager out-of-memory risk
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Parse only captured tables to accelerate reads
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

For sensitive fields such as username and password, use the ${secret_values.<key>} syntax and manage credentials with Variable Management.

(Optional) Click Depth Check to validate syntax, network connectivity, and access permissions.