Realtime Compute for Apache Flink:Complex use cases for Flink CDC jobs

Hot-synchronize new empty tables

To synchronize newly created empty tables in real time during the incremental phase without a restart, set the scan.binlog.newly-added-table.enabled parameter. This method is recommended as it avoids restarting the job.

For example, a data ingestion job is synchronizing all tables from the MySQL dlf_test database. A new empty table named products is created in the source. To synchronize this new table without restarting the job, set scan.binlog.newly-added-table.enabled: true in the job configuration as shown below:

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize data from tables newly 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 prevent potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

With this configuration, the job automatically creates all new tables from the dlf_test database in the destination.

Synchronize tables with historical data

Assume a MySQL database already contains the customers and products tables. However, you initially configure the job to synchronize only the customers table:

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.customers
  server-id: 8601-8604
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

After the job has been running for a while, if you need to synchronize all tables from the database, including their historical data, you must restart the job. Follow these steps:

1. Stop the job with a savepoint.

2. Modify the tables parameter in the MySQL source configuration to match all tables. Then, remove the scan.binlog.newly-added-table.enabled parameter if it exists and add scan.newly-added-table.enabled.

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize full and incremental data for newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

3. Restart the job from the savepoint.

For example, to synchronize all tables in the dlf_test MySQL database except for a table named products_tmp, use the following configuration:

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  # (Optional) Exclude tables that you do not want to synchronize.
  tables.exclude: dlf_test.products_tmp
  server-id: 8601-8604
  # (Optional) Synchronize data from tables newly 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 prevent potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

A Flink CDC data ingestion job with this configuration automatically creates all tables from the dlf_test database in the destination, except for the products_tmp table. The job keeps the schemas and data of the synchronized tables updated in real time.

Add metadata columns

When writing data, you can use the transform module to add metadata columns. For example, the following configuration adds the table name, operation time, and operation type to the downstream table. For more details, see Transform module.

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize full and incremental data for newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # Include the operation time as metadata.
  metadata-column.include-list: op_ts
transform:
  - source-table: dlf_test.customers
    projection: __schema_name__ || '.' || __table_name__  as identifier, op_ts, __data_event_type__ as op, *
    # (Optional) Modify the primary key.
    primary-keys: id,identifier
    description: add identifier, op_ts and op
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

A source table contains all change data event types. To implement a soft delete by converting DELETE operations to INSERTs in the downstream table, add the converter-after-transform: SOFT_DELETE configuration in the transform module.

Add computed columns

You can use the transform module to add computed columns when writing data. For example, the following configuration creates a dt field by transforming the created_at field and uses it as a partition key for the downstream table.

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize full and incremental data for newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # Include the operation time as metadata.
  metadata-column.include-list: op_ts
transform:
  - source-table: dlf_test.customers
    projection: DATE_FORMAT(created_at, 'yyyyMMdd') as dt, *
    # (Optional) Set partition keys.
    partition-keys: dt
    description: add dt
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Merge sharded tables

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize full and incremental data for newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  # Merge all tables from the dlf_test database with names matching the pattern product_[0-9]+ into the dlf.products table.
  - source-table: dlf_test.product_[0-9]+
    sink-table: dlf.products
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Database synchronization

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Synchronize full and incremental data for newly added tables.
  scan.newly-added-table.enabled: true
  # (Optional) Synchronize table and column 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 a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  # Rename all tables from the dlf_test database by synchronizing them to the dlf database and prefixing the source table names with ods_.
  - source-table: dlf_test.\.*
    sink-table: dlf.ods_<>
    replace-symbol: <>
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

O&M page configuration

On the job's O&M page, you can specify the start time for the source table when performing a stateless start.

This configuration is supported for MySQL and Kafka sources. Enable the corresponding switch and set a specific start date and time.

Job parameter configuration

In the job definition, you can set parameters to specify the start time for the source table.

For example, with a MySQL source, you can set scan.startup.mode: timestamp in the job configuration to start from a specific timestamp. The following is an example configuration:

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: dlf_test.\.*
  server-id: 8601-8604
  # (Optional) Start the job in timestamp mode.
  scan.startup.mode: timestamp
  # Specify the startup timestamp in this mode.
  scan.startup.timestamp-millis: 1667232000000
  # (Optional) Synchronize data from tables newly 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 prevent potential TaskManager OutOfMemory issues.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable a parsing filter to accelerate reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Specify a commit user. We recommend using a different user for each job to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true