Definition #

Transform module helps users delete and expand data columns based on the data columns in the table.
What’s more, it also helps users filter some unnecessary data during the synchronization process.

Parameters #

To describe a transform rule, the following parameters can be used:

Multiple rules can be declared in one single pipeline YAML file.

converter-after-transform #

converter-after-transform is used to change the DataChangeEvent after other transform. The available values of this options are as follows.

• SOFT_DELETE: The delete event will be converted as an insert event. This converter should be used together with the metadata __data_event_type__. Then you can implement the soft delete.

For example, the following transform will not delete data when the delete event happens. Instead it will update the column op_type to -D in sink and transform it to an insert record.

transform:
  - source-table: \.*.\.*
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

Metadata Fields #

Fields definition #

There are some hidden columns used to access metadata information. They will only take effect when explicitly referenced in the transform rules.

Besides these fields, pipeline connectors could parse more metadata and put them in the meta map of the DataChangeEvent. These metadata could be accessed in the transform module. For example, MySQL pipeline connector could parse op_ts and use it in the transform module.

source:
  type: mysql
  hostname: localhost
  port: 3306
  username: testuser
  password: testpwd
  tables: testdb.customer
  server-id: 5400-5404
  server-time-zone: UTC
  metadata.list: op_ts
  
transform:
  - source-table: testdb.customer
    projection: \*, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ AS identifier_name, __data_event_type__ AS type, op_ts AS opts
  
sink:
  type: values

Metadata relationship #

Functions #

Flink CDC uses Calcite to parse expressions and Janino script to evaluate expressions with function call.

Comparison Functions #

Logical Functions #

Arithmetic Functions #

String Functions #

Temporal Functions #

Conditional Functions #

Casting Functions #

You can use CAST( <EXPR> AS <T> ) syntax to convert any valid expression <EXPR> to a specific type <T>. Possible conversion paths are:

Variant Functions #

Struct Functions #

Struct functions are used to access elements in ARRAY, MAP, ROW, and VARIANT types using subscript syntax.

Example #

Add computed columns #

Evaluation expressions can be used to generate new columns. For example, if we want to append two computed columns based on the table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name, localtimestamp as new_timestamp
    description: append calculated columns based on source table

Reference metadata columns #

We may reference metadata column in projection expressions. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, __namespace_name__ || '.' || __schema_name__ || '.' || __table_name__ identifier_name
    description: access metadata columns from source table

Use wildcard character to project all fields #

A wildcard character (*) can be used to reference all fields in a table. For example, given two tables web_order and app_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: \*, UPPER(product_name) as product_name
    description: project fields with wildcard character from source table
  - source-table: mydb.app_order
    projection: UPPER(product_name) as product_name, *
    description: project fields with wildcard character from source table

Notice: When * character presents at the beginning of expressions, an escaping backslash is required.

Add filter rule #

Use reference columns when adding filtering rules to the table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    filter: id > 10 AND order_id > 100
    description: filtering rows from source table

Computed columns can be used in filtering conditions, too. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(province) as new_province 
    filter: new_province = 'SHANGHAI'
    description: filtering rows based on computed columns

Reassign primary key #

We can reassign the primary key in transform rules. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id
    primary-keys: order_id
    description: reassign primary key example

Composite primary keys are also supported:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    primary-keys: order_id, product_name
    description: reassign composite primary keys example

Notice that primary key columns will be attributed as NOT NULL in the downstream table, so you should ensure that no NULL value will be assigned to these columns.

Reassign partition key #

We can reassign the partition key in transform rules. For example, given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    partition-keys: product_name
    description: reassign partition key example

Specify table creation configuration #

Extra options can be defined in a transform rule, and will be applied when creating downstream tables. Given a table web_order in the database mydb, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id, UPPER(product_name) as product_name
    table-options: comment=web order
    description: auto creating table options example

Tips: The format of table-options is key1=value1,key2=value2. If option values contain commas or other special characters, you can specify a custom delimiter using table-options.delimiter (such as ;, |, $, etc.):

transform:
  - source-table: mydb.web_order
    table-options: sequence.field=gxsj,jjsj;file-index.bloom-filter.columns=jjdbh
    table-options.delimiter: ";"

Classification mapping #

If a table hits ultiple transform rules, only the first matched transform rule will apply. For example, we may define a transform rule as follows:

transform:
  - source-table: mydb.web_order
    projection: id, order_id
    filter: id > 1001
  - source-table: mydb.\.*
    projection: \*, 'fallback' AS FALLBACK

Here, though mydb.web_order matches the second rule (mydb.\.*), it will not fall through the next rule as it has been handled in the first rule.

User-defined Functions #

User-defined functions (UDFs) can be used in transform rules.

Classes could be used as a UDF if:

• implements org.apache.flink.cdc.common.udf.UserDefinedFunction interface
• has a public constructor with no parameters
• has at least one public method named eval

It may also:

• overrides getReturnType method to indicate its return CDC type
• overrides open and close method to do some initialization and cleanup work

For example, this is a valid UDF class:

public class AddOneFunctionClass implements UserDefinedFunction {
    
    public Object eval(Integer num) {
        return num + 1;
    }
    
    @Override
    public DataType getReturnType() {
        return DataTypes.INT();
    }
    
    @Override
    public void open() throws Exception {
        // ...
    }

    @Override
    public void close() throws Exception {
        // ...
    }
}

To ease the migration from Flink SQL to Flink CDC, a Flink ScalarFunction could also be used as a transform UDF, with some limitations:

• ScalarFunction which has a constructor with parameters is not supported.
• Flink-style type hint in ScalarFunction will be ignored.
• open / close lifecycle hooks will not be invoked.

UDF classes could be registered by adding a user-defined-function block:

pipeline:
  user-defined-function:
    - name: addone
      classpath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass
    - name: format
      classpath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass

Notice that given classpath must be fully-qualified, and corresponding jar files must be included in Flink /lib folder, or be passed with flink-cdc.sh --jar option.

UDF Options #

You can pass extra options to UDFs by adding an options block. These options will be available in the open method through UserDefinedFunctionContext.configuration():

pipeline:
  user-defined-function:
    - name: query_redis
      classpath: com.example.flink.cdc.udf.RedisQueryFunction
      options:
        hostname: localhost
        port: "6379"
        cache.enabled: "true"

And in your UDF implementation, you can access these options by defining ConfigOption instances:

import org.apache.flink.cdc.common.configuration.ConfigOption;
import org.apache.flink.cdc.common.configuration.ConfigOptions;

public class RedisQueryFunction implements UserDefinedFunction {
    private static final ConfigOption<String> HOSTNAME =
        ConfigOptions.key("hostname").stringType().noDefaultValue();
    private static final ConfigOption<Integer> PORT =
        ConfigOptions.key("port").intType().defaultValue(6379);

    private String hostname;
    private int port;
    
    @Override
    public void open(UserDefinedFunctionContext context) throws Exception {
        hostname = context.configuration().get(HOSTNAME);
        port = context.configuration().get(PORT);
        // Initialize your connection here...
    }
    
    public Object eval(String key) {
        // Query Redis using hostname and port...
    }
}

The options field is optional. If not specified, an empty configuration will be passed to the UDF.

After being correctly registered, UDFs could be used in both projection and filter expressions, just like built-in functions:

transform:
  - source-table: db.\.*
    projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
    filter: inc(id) < 100

Embedding AI Model #

Embedding AI Model can be used in transform rules. To use Embedding AI Model, you need to download the jar of build-in model, and then add --jar {$BUILT_IN_MODEL_PATH} to your flink-cdc.sh command.

How to define a Embedding AI Model:

pipeline:
  model:
    - model-name: CHAT
      class-name: OpenAIChatModel
      openai.model: text-embedding-3-small
      openai.host: https://xxxx
      openai.apikey: abcd1234
      openai.chat.prompt: please summary this
    - model-name: GET_EMBEDDING
      class-name: OpenAIEmbeddingModel
      openai.model: text-embedding-3-small
      openai.host: https://xxxx
      openai.apikey: abcd1234

Note:

• model-name is a common required parameter for all support models, which represent the function name called in projection or filter.
• class-name is a common required parameter for all support models, available values can be found in All Support models.
• openai.model, openai.host, openai.apiKey and openai.chat.prompt is option parameters that defined in specific model.

How to use a Embedding AI Model:

transform:
  - source-table: db.\.*
    projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb, CHAT(page) as summary"
    filter: inc(id) < 100
pipeline:
  model:
    - model-name: CHAT
      class-name: OpenAIChatModel
      openai.model: gpt-4o-mini
      openai.host: http://langchain4j.dev/demo/openai/v1
      openai.apikey: demo
      openai.chat.prompt: please summary this
    - model-name: GET_EMBEDDING
      class-name: OpenAIEmbeddingModel
      openai.model: text-embedding-3-small
      openai.host: http://langchain4j.dev/demo/openai/v1
      openai.apikey: demo

Here, GET_EMBEDDING is defined though model-name in pipeline.

All Support models #

The following built-in models are provided:

OpenAIChatModel #

OpenAIEmbeddingModel #