Connect Flink CDC to DLF catalog via Iceberg REST | Realtime Compute for Apache Flink - Data Lake Formation

Prerequisites

Before you begin, make sure you have:

A fully managed Flink workspace. See Activate Realtime Compute for Apache Flink.
The Flink workspace and DLF in the same region.
The Virtual Private Cloud (VPC) where the Flink workspace resides added to the DLF whitelist. See Configure a VPC whitelist.

Limitations

Iceberg REST connectivity to DLF requires Realtime Compute for Apache Flink engine VVR 11.6.0 or later.

Register the DLF catalog in Flink

Registering a catalog in Flink creates a mapping to your DLF catalog. Creating or deleting the catalog in Flink does not affect the actual data in DLF. All tables created in the DLF catalog through Iceberg REST are Iceberg tables.

Log on to the Realtime Compute for Apache Flink Management Console.
In the Actions column of your workspace, click Console.
In the left navigation pane, click Development > Scripts.

Create a new script, then paste the following SQL statement into the SQL editor.

CREATE CATALOG `catalog_name`
 WITH (
    'type' = 'iceberg',
    'catalog-type' = 'rest',
    'uri' = 'http://{region-id}-vpc.dlf.aliyuncs.com/iceberg',
    'warehouse' = 'iceberg_test',
    'rest.signing-region' = '{region-id}',
    'io-impl' = 'org.apache.iceberg.rest.DlfFileIO'
);

Replace {region-id} with the region ID of your DLF catalog, for example cn-hangzhou or ap-southeast-1. For all supported regions and their endpoint values, see Regions and endpoints.

In the lower-right corner, click Environment, select a session cluster running VVR 11.2.0 or later, and run the SQL statement.

The following table describes the configuration options.

Option	Description	Required	Example
`type`	The catalog type. Set to `iceberg`.	Yes	`iceberg`
`catalog-type`	The catalog type. Set to `rest`.	Yes	`rest`
`token.provider`	The token provider for DLF authentication. Set to `dlf`.	Yes	`dlf`
`uri`	The Iceberg REST endpoint for your DLF catalog. Use the format `http://{region-id}-vpc.dlf.aliyuncs.com/iceberg`. For region-specific values, see Regions and endpoints.	Yes	`http://ap-southeast-1-vpc.dlf.aliyuncs.com/iceberg`
`warehouse`	The name of your DLF catalog.	Yes	`iceberg_test`
`rest.signing-region`	The region ID of your DLF catalog. For region IDs, see Regions and endpoints.	Yes	`ap-southeast-1`
`io-impl`	The FileIO implementation for DLF. Set to `org.apache.iceberg.rest.DlfFileIO`.	Yes	`org.apache.iceberg.rest.DlfFileIO`

Configure Flink CDC to use the catalog

Create a data ingestion job: Develop a Flink CDC data ingestion job.

If you already have a Flink Catalog mapping, Reuse an existing Catalog to obtain connection information and add this Sink configuration:

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

Configuration examples

Common data sync patterns using Flink CDC YAML jobs:

Sync a full MySQL database to DLF

This job syncs an entire MySQL database to DLF:

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync tables created during incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded chunks to prevent TaskManager OOM errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Speed up reads by parsing only matched tables.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

Note

Recommended optional parameters for MySQL sources (MySQL):

Parameter: scan.binlog.newly-added-table.enabled

Function: Syncs tables created during incremental phase.
Parameter: include-comments.enabled

Function: Syncs table and field comments.
Parameter: scan.incremental.snapshot.unbounded-chunk-first.enabled

Function: Prevents TaskManager OOM errors.
Parameter: scan.only.deserialize.captured.tables.changelog.enabled: true

Function: Speeds up reads by parsing only matched tables.

Write to a partitioned DLF table

Specify partition keys with the partition-keys parameter (Flink CDC data ingestion job development reference):

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync tables created during incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded chunks to prevent TaskManager OOM errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Speed up reads by parsing only matched tables.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

transform:
  - source-table: mysql_test.tbl1
    # (Optional) Set partition keys.
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

Write to an append-only DLF table

To implement soft deletes (convert DELETE operations to INSERTs downstream), configure the job as follows (Flink CDC data ingestion job development reference):

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync tables created during incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded chunks to prevent TaskManager OOM errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Speed up reads by parsing only matched tables.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

transform:
  - source-table: mysql_test.tbl1
    # (Optional) Set partition keys.
    partition-keys: id,pt
    # (Optional) Implement soft delete.
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE
  - source-table: mysql_test.tbl2
    # (Optional) Set partition keys.
    partition-keys: id,pt
    # (Optional) Implement soft delete.
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

Note

Adding __data_event_type__ to the projection writes the change event type as a new downstream field. Setting converter-after-transform to SOFT_DELETE converts DELETEs to INSERTs, recording all change events. Flink CDC data ingestion job development reference.

Sync real-time Kafka CDC data to DLF

This example syncs change data for two tables (customers and products) from a Kafka inventory topic in Debezium JSON format to DLF:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

# Debezium JSON lacks primary key info. Add it manually.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Note

The Kafka source supports canal-json, debezium-json (default), and json formats.
When using debezium-json, manually add a primary key using a transform rule, since Debezium JSON messages don't contain primary key information:
```
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
```
If a single table's data spans multiple partitions, or tables across partitions need merging, set debezium-json.distributed-tables or canal-json.distributed-tables to true.
The Kafka source supports multiple schema inference strategies via the schema.inference.strategy parameter. Kafka.

Sync real-time Kafka logs to DLF

For custom JSON data in Kafka, Flink CDC handles data type inference, schema inference, and schema evolution automatically.

This example syncs a single JSON log table from the inventory topic to DLF:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # (Optional) Recursively flatten nested columns in JSON data.
  json.infer-schema.flatten-nested-columns.enable: true
  # (Optional) Skip first 100 parsing errors. Job fails if errors exceed 100.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

# Add primary key to table.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

# Write all inventory topic data to test_database.inventory.
route:
  - source-table: inventory
    sink-table: test_database.inventory

pipeline:
  # (Optional) Log dirty data that causes processing exceptions.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

This example syncs multiple JSON log tables from the inventory topic, using databaseName and tableName fields to identify tables:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: json
  # (Optional) Recursively flatten nested columns in JSON data.
  json.infer-schema.flatten-nested-columns.enable: true
  # Use databaseName and tableName field values as database and table names.
  json.decode.parser-table-id.fields: databaseName,tableName
  # (Optional) Skip first 100 parsing errors. Job fails if errors exceed 100.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: iceberg
  using.built-in-catalog: catalog_name

# Add primary key to tables.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

# Write ods.inventory, ods.customer, and ods.user to test_database.inventory, test_database.customer, and test_database.user respectively.
route:
  - source-table: ods.inventory
    sink-table: test_database.inventory
  - source-table: ods.customer
    sink-table: test_database.customer
  - source-table: ods.user
    sink-table: test_database.user

pipeline:
  # (Optional) Log dirty data that causes processing exceptions.
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

JSON schema inference and evolution strategies: Schema parsing and change synchronization strategies.

Full configuration reference: Flink CDC data ingestion job development reference.