Flink CDC source and sink modules-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

This topic describes the source and sink modules for Flink Change Data Capture (CDC) data ingestion jobs and lists the supported connectors.

Supported connectors

Connector	Supported type
Connector	Source	Sink
MySQL Note Supports connections to ApsaraDB RDS for MySQL, PolarDB for MySQL, and self-managed MySQL.	√	×
Paimon	×	√
Fluss (Public Preview)	×	√ Note Supported only in Ververica Runtime (VVR) 11.4.0 and later.
Kafka	√ Note Supported only in VVR 8.0.10 and later.	√
Upsert Kafka	×	√
StarRocks	×	√
Hologres	×	√
Simple Log Service	√ Note Supported only in VVR 11.1 and later.	×
MongoDB	√ Note Supported only in VVR 11.2 and later.	×
MaxCompute	×	√ Note Supported only in VVR 11.1 and later.
SelectDB	×	√ Note Supported only in VVR 11.1 and later.
PostgreSQL CDC (Public Preview)	√ Note Supported only in VVR 11.4 and later.	×
Print	×	√
Iceberg	×	√ Note Supported only in VVR 11.6 and later.

Connector configuration

You can configure parameters for source and sink connectors in a Flink CDC data ingestion job. For details about the supported connectors and their parameters, see the following sections.

# Source module
source:
  type: mysql # Or another connector identifier
  name: MySQL Source
  # Other parameters. Use key: value pairs.

# Sink module
sink:
  type: paimon # Or another connector identifier
  name: Paimon Sink
  # Other parameters. Use key: value pairs.

General configuration

Parameter	Description	Required	Type	Default	Remarks
type	The connector type for the source or sink.	Yes	String	None	None
name	The name of the node.	No	String	None	None
using.built-in-catalog	Reuses a built-in catalog.	No	String	None	Available in VVR 11.5 and later. This parameter is supported only by specific connectors. For a list of supported connectors, see Reuse connection information from a catalog.

Reuse connection information from a catalog

Starting with VVR 11.5, you can directly reference a built-in catalog created on the Data Management page for your Flink CDC data ingestion job. This reuses connection properties such as URL, username, and password, simplifying manual configuration.

Syntax

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  
sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog

You can use the using.built-in-catalog parameter in the source and sink modules to reference an existing built-in catalog.

For example, in the example above, the Catalog metadata for mysql_rds_catalog already includes required parameters such as hostname, username, and password. Therefore, you do not need to provide these parameters again in the YAML job.

Usage notes

The following connectors support reusing connection information from a catalog:

MySQL (source)
Kafka (source)
Upsert Kafka (sink)
StarRocks (sink)
Hologres (sink)
Paimon (sink)
Simple Log Service (source)
Iceberg (sink)

Note

Catalog parameters that are incompatible with the CDC YAML configuration are ignored. For more information, see the parameter list for each connector.

Source configuration

Parameter	Description	Required	Type	Default	Remarks
source-expand	Specifies the distribution strategy for data emitted from the source.	No	See the syntax section for configuration details.	None	Available in VVR 11.6 and later.

source-expand

The source-expand parameter distributes or replicates data before the Transform and Route modules process it.

Syntax

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # Replicate mydb.orders into three tables, each processed differently before writing to the sink.
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# Apply different transformations to each of the three expanded tables
transform:
  # DWD layer: Retain all fields, add a calculated column
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # DWS layer: Retain only key fields, filter out small orders
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # ADS layer: Retain only fields needed for reporting, filter out canceled orders
  - source-table: ads.orders_report
    projection: order_id, user_id, amount, TO_UPPER(order_status) as status
    filter: order_status <> 'CANCELLED'
    primary-keys: order_id

# Route the three expanded tables to different sink tables
route:
  - source-table: dwd.orders_full
    sink-table: starrocks_dwd.orders_full_detail
  - source-table: dws.orders_summary
    sink-table: starrocks_dws.orders_summary
  - source-table: ads.orders_report
    sink-table: starrocks_ads.orders_report

sink:
  type: starrocks
  name: sink-starrocks
  jdbc-url: jdbc:mysql://localhost:9030
  load-url: localhost:8030
  username: root
  password: pass

As shown in the example, this configuration replicates data from the mydb.orders table to three logical tables: dwd.orders_full, dws.orders_summary, and ads.orders_report. Each of these tables is then processed differently before being written to a separate sink table.

Usage notes

Available in VVR 11.6 and later.
By default, the original table is not retained after data distribution. In the syntax example, the mydb.orders table is removed from the data stream. To retain the original table, add it to the output-table list.
```
source-expand:
  - input-table: mydb.orders
    output-table: [ mydb.orders, db1.orders, db2.orders ]
```
The input-table and output-table parameters do not support regular expressions.

Sink configuration

Parameter	Description	Required	Type	Default	Remarks
include.schema.changes	Specifies which types of schema changes to apply.	No	List<String>	None	By default, all schema changes are synchronized.
exclude.schema.changes	Specifies which types of schema changes to exclude.	No	List<String>	None	Has a higher priority than `include.schema.changes`.

For more information about how to use include.schema.changes and exclude.schema.changes, see Schema Change Synchronization Configuration.