Develop a Flink CDC data ingestion job

In the Actions column for the target workspace, click Console.

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

Click and then click New Draft with Template.

Select a data synchronization template.

Currently, only the MySQL to StarRocks, MySQL to Paimon, and MySQL to Hologres templates are available.

Enter the job information, such as draft name, location, and engine version, and then click OK.

Configure the source and sink information for the Flink CDC job.

For parameter configuration details, see the documentation for the relevant connectors.

In the Actions column for the target workspace, click Console.

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

Click , and then select New Draft. Enter the name and engine version, and then click Create.

Paste the YAML code from your open-source Flink CDC job into the editor.

(Optional) Click Validate.

This option checks for syntax errors, network connectivity issues, and permission problems.

In the Actions column for the target workspace, click Console.

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

Click , and then select New Draft. Enter the name and engine version, and then click Create.

Configure the Flink CDC job using YAML. Example:

# Required
source:
  # The type of the data source.
  type: <Replace with your source connector type>
  # The configuration for the data source. For details about configuration items, see the documentation for the corresponding connector.
  ...
# Required
sink:
  # The type of the sink.
  type: <Replace with your sink connector type>
  # The configuration for the sink. For details about configuration items, see the documentation for the corresponding connector.
  ...
# Optional
transform:
  # A transform rule for the flink_test.customers table.
  - source-table: flink_test.customers
    # Projection configuration. Specifies the columns to synchronize and performs data transformations.
    projection: id, username, UPPER(username) as username1, age, (age + 1) as age1, test_col1, __schema_name__ || '.' || __table_name__ identifier_name
    # Filter condition. Synchronizes only data where id is greater than 10.
    filter: id > 10
    # Description of the transform rule.
    description: append calculated columns based on source table
# Optional
route:
  # A route rule that specifies the mapping between the source table and the sink table.
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # Description of the route rule.
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # Description of the route rule.
    description: sync customers_suffix table
# Optional
pipeline:
  # The name of the job.
  name: MySQL to Hologres Pipeline

The following table describes the code blocks.

For more information about the syntax and configuration of each module, see Reference for Flink CDC data ingestion jobs.

The following code provides an example of synchronizing all tables from the app_db database in MySQL to a database in Hologres.

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404
  # (Optional) Synchronize data from tables newly created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded chunks to prevent potential TaskManager OutOfMemory issues.
  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: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}
pipeline:
  name: Sync MySQL Database to Hologres

(Optional) Click Validate.

This option checks for syntax errors, network connectivity issues, and permission problems.