Realtime Compute for Apache Flink:Develop a Flink CDC data ingestion job (public preview)

Click Console in the Actions column of the target workspace.

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

Click , and then click Create From Template.

Select a data synchronization template.

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

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

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

For more information about parameter settings, see the documentation for the corresponding connector.

Click Console in the Actions column of the target workspace.

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

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

Copy the Flink CDC job from the open source community.

(Optional) Click Depth Check.

You can check the syntax, network connectivity, and access permissions.

Click Console in the Actions column of the target workspace.

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

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

Configure the Flink CDC job.

# Required
source:
  # Data source type
  type: <Replace with your source connector type>
  # Data source configurations. For more information about the configuration items, see the documentation for the corresponding connector.
  ...

# Required
sink:
  # Destination type
  type: <Replace with your destination connector type>
  # Destination configurations. For more information about the configuration items, see the documentation for the corresponding connector.
  ...

# Optional
transform:
  # Transformation rule for the flink_test.customers table
  - source-table: flink_test.customers
    # Projection configuration. Specifies the columns to synchronize and performs data transformation.
    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 transformation rule
    description: append calculated columns based on source table

# Optional
route:
  # Routing rule that specifies the mapping between source and destination tables.
  - source-table: flink_test.customers
    sink-table: db.customers_o
    # Description of the routing rule
    description: sync customers table
  - source-table: flink_test.customers_suffix
    sink-table: db.customers_s
    # Description of the routing rule
    description: sync customers_suffix table

#Optional
pipeline:
  # Job name
  name: MySQL to Hologres Pipeline

The code blocks are described as follows.

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

The following code provides an example of how to synchronize 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 newly created tables 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 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

(Optional) Click Depth Check.

You can check the syntax, network connectivity, and access permissions.