Develop a data ingestion YAML job - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink uses Flink CDC to perform data ingestion. You can develop YAML jobs to efficiently synchronize data from a source to a sink. This topic describes how to develop a Flink CDC data ingestion job.

Background information

Flink CDC data ingestion uses Flink CDC to simplify data integration. By using YAML to define complex ETL processes that are automatically converted into Flink runtime logic, you can implement features like full database sync, sharded table sync, schema evolution, and computed columns. This approach greatly simplifies the data integration process and improves its efficiency and reliability.

Advantages of Flink CDC

In Realtime Compute for Apache Flink, you can develop a Flink CDC data ingestion job, a SQL job, or a DataStream job to perform data synchronization. The following sections describe the advantages of Flink CDC data ingestion jobs over the other two development methods.

Flink CDC vs. Flink SQL

Flink CDC data ingestion jobs and Flink SQL jobs transmit different types of data:

SQL jobs transmit RowData, where each row has a change type: insert (+I), update before (-U), update after (+U), or delete (-D).
Flink CDC jobs use SchemaChangeEvent to transmit schema evolution information, such as table creation, column addition, or table truncation. They use DataChangeEvent to transmit data changes such as insert, update, and delete. An update message contains both the before and after states of a row, enabling you to write the original change data to the sink.

The following table lists the advantages of Flink CDC data ingestion jobs over SQL jobs.

Flink CDC	Flink SQL
Automatically discovers schemas and supports full database sync	Requires manual `CREATE TABLE` and `INSERT` statements
Supports multiple schema evolution policies	Does not support schema evolution
Preserves the original changelog	Disrupts the original changelog structure
Reads from and writes to multiple tables	Reads from and writes to a single table

Compared to CTAS or CDAS statements, Flink CDC jobs offer the following advantages:

Immediately synchronizes upstream schema evolution without waiting for new data writes to trigger the process.
Preserves the original changelog, and UPDATE messages are not split.
Synchronizes more types of schema evolution, such as TRUNCATE TABLE and DROP TABLE.
Supports table mappings to flexibly define sink table names.
Supports flexible, user-configurable schema evolution behaviors.
Supports data filtering by using WHERE clauses.
Supports column pruning.

Flink CDC vs. Flink DataStream

The following table lists the advantages of Flink CDC data ingestion jobs over DataStream jobs.

Flink CDC	Flink DataStream
Designed for users at all skill levels, not just experts	Requires expertise in Java and distributed systems
Hides low-level details and simplifies development	Requires familiarity with the Flink framework
The YAML format is easy to understand and learn	Requires tools like Maven to manage dependencies
Existing jobs are easy to reuse	Existing code is difficult to reuse

Limitations

Use Ververica Runtime (VVR) 11.1 or later to develop Flink CDC data ingestion jobs. If you need to use a VVR 8.x version, use VVR 8.0.11.
Each job supports only one source and one sink. To read from multiple sources or write to multiple sinks, you must create multiple Flink CDC jobs.
Flink CDC jobs cannot be deployed to a session cluster.
Flink CDC jobs do not support automatic tuning.

Flink CDC data ingestion connectors

See Supported connectors for a list of connectors supported as sources and sinks for Flink CDC data ingestion.

Create a Flink CDC data ingestion job

From a template

Log on to the Realtime Compute for Apache Flink console.
In the Actions column of the target workspace, click Console.
In the left-side navigation pane, choose Development > Data Ingestion.
Click and then click New from Template.
Select a data synchronization template.
Templates are currently available for MySQL to StarRocks, MySQL to Paimon, and MySQL to Hologres.
Enter the job information, including the job name, storage location, and engine version, and then click OK.
Configure the source and sink information for the Flink CDC job.
See the documentation for the corresponding connector for parameter configuration details.

From CTAS/CDAS

Important

If a job contains multiple CTAS or CDAS statements, Flink detects and converts only the first one.
Due to differences in built-in function support between Flink SQL and Flink CDC, the generated transform rules may not work as-is. You must review and adjust them as needed.
If the source is MySQL and the original CTAS/CDAS job is still running, you must adjust the source server-id of the Flink CDC data ingestion job to avoid conflicts.

Log on to the Realtime Compute for Apache Flink console.
In the Actions column of the target workspace, click Console.
In the left-side navigation pane, choose Development > Data Ingestion.
Click and then click New from CTAS/CDAS Job. Select the target CTAS or CDAS job and click OK.
On the selection page, only valid CTAS and CDAS jobs are displayed. It does not display regular ETL jobs or drafts with syntax errors.
Enter the job information, including the job name, storage location, and engine version, and then click OK.

From open source

Log on to the Realtime Compute for Apache Flink console.
In the Actions column of the target workspace, click Console.
In the left-side navigation pane, choose Development > Data Ingestion.
Click , select New Data Ingestion Draft, enter the File Name and Engine Version, and then click Create.
Copy the code of the open source Flink CDC job.
(Optional) Click Validate.
You can validate the syntax, network connectivity, and access permissions.

From scratch

Log on to the Realtime Compute for Apache Flink console.
In the Actions column of the target workspace, click Console.
In the left-side navigation pane, choose Development > Data Ingestion.
Click , select New Data Ingestion Draft, enter the File Name and Engine Version, and then click Create.

Configure the Flink CDC job.

# Required
source:
  # Source connector type
  type: <Replace with your source connector type>
  # Source configurations. For details, see the documentation for the corresponding connector.
  ...

# Required
sink:
  # Sink connector type
  type: <Replace with your sink connector type>
  # Sink configurations. For details, see the documentation for the corresponding connector.
  ...

# Optional
transform:
  # Transformation rule for the flink_test.customers table
  - source-table: flink_test.customers
    # Projection settings. 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. Specifies the mapping between source tables and sink 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

Note

In a Flink CDC job, the Key and Value must be separated by a space in the format Key: Value.

The following table describes the code blocks.

Required	Module	Description
Required	source	The start of the data pipeline. Flink CDC captures change data from the source. Note Currently, MySQL is the only supported source. For details about specific configuration items, see MySQL. You can use variables to manage sensitive information. For more information, see Variable Management.
Required	sink	The end of the data pipeline. Flink CDC transmits the captured data changes to the sink. Note For the supported sink systems, see Flink CDC data ingestion connectors. For details about sink configuration items, see the documentation for the corresponding connector. You can use variables to manage sensitive information. For more information, see Variable Management.
Optional	pipeline (data pipeline)	Defines basic configurations for the entire data pipeline job, such as the pipeline name.
	transform (data transformation)	Specifies data transformation rules. Transformations operate on data as it flows through the Flink pipeline. Supported features include ETL processing, filtering with `WHERE` clauses, column pruning, and computed columns. Use the `transform` module to transform raw change data from Flink CDC to fit a specific downstream system.
	route	If this module is not configured, the job performs a full database or specified table synchronization. In some cases, you may need to send captured change data to different destinations based on specific rules. The routing mechanism allows you to flexibly specify the mapping relationship between the source and the sink to send data to different sinks.

For details on the syntax and configuration of each module, see Reference for developing 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 tables created during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize dispatching unbounded chunks to avoid potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parse filtering to speed up 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 Validate.
You can validate the syntax, network connectivity, and access permissions.

Realtime Compute for Apache Flink:Develop a Flink CDC data ingestion job