定义
Transform 模块允许用户根据数据列对数据列进行删除和扩展。 更进一步的，这也也可以帮助用户在同步过程中过滤一些不需要的数据。
参数
为了定义一个 transform 规则，可以使用以下参数：
| 参数 | 含义 | 是否必填 |
|---|---|---|
| source-table | 源表 ID，支持正则表达式 | 必填 |
| projection | 投影规则，支持类似 SQL 中 SELECT 子句的语法 | 可选 |
| filter | 过滤规则，支持类似 SQL 中 WHERE 子句的语法 | 可选 |
| primary-keys | 目标表主键，以逗号分隔 | 可选 |
| partition-keys | 目标表分区键，以逗号分隔 | 可选 |
| table-options | 用于配置自动建表时的建表语句 | 可选 |
| table-options.delimiter | 多个表属性的分隔符，默认值为 , | 可选 |
| converter-after-transform | 用于在 transform 处理后添加转换器来修改 DataChangeEvent | 可选 |
| description | Transform 规则描述 | 可选 |
多个 transform 规则可以声明在一个 pipeline YAML 文件中。
converter-after-transform
converter-after-transform 用于在其他 transform 处理之后再修改 DataChangeEvent。可用的值如下：
- SOFT_DELETE: 删除事件会被转换成插入事件。这个转换器应该与元数据
__data_event_type__一起使用，然后你可以实现软删除。
举个例子，下面的 transform 将不会删除数据，而是更新列 op_type 为 -D，然后转换成插入记录。
transform:
- source-table: \.*.\.*
projection: \*, __data_event_type__ AS op_type
converter-after-transform: SOFT_DELETE
元数据字段
字段定义
有一些隐藏列可以用来访问元数据信息。它们仅在 transform 规则中显式引用时才会生效。
| 字段 | 数据类型 | 描述 |
|---|---|---|
| namespace_name | String | 包含该行数据的命名空间名称。 |
| schema_name | String | 包含该行数据的 schema 名称。 |
| table_name | String | 包含该行数据的表名称。 |
| data_event_type | String | 数据变更事件的操作类型。 |
除了以上的元数据字段，连接器也可以解析额外的元数据并放入 DataChangeEvent 的 meta map 中，这些元数据也可以在 transform 中使用。
例如如下的作业，MySQL 连接器可以解析 op_ts 元数据并在 transform 中使用。
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
元数据关系
| 类型 | Namespace | SchemaName | Table |
|---|---|---|---|
| JDBC | Catalog | Schema | Table |
| Debezium | Catalog | Schema | Table |
| MySQL | Database | - | Table |
| Postgres | Database | Schema | Table |
| Oracle | - | Schema | Table |
| Microsoft SQL Server | Database | Schema | Table |
| StarRocks | Database | - | Table |
| Doris | Database | - | Table |
函数
Flink CDC 使用 Calcite 来解析表达式并且使用 Janino script 来执行表达式。
比较函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| value1 = value2 | valueEquals(value1, value2) | 如果 value1 等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 <> value2 | !valueEquals(value1, value2) | 如果 value1 不等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 > value2 | greaterThan(value1, value2) | 如果 value1 大于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 >= value2 | greaterThanOrEqual(value1, value2) | 如果 value1 大于等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 < value2 | lessThan(value1, value2) | 如果 value1 小于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value1 <= value2 | lessThanOrEqual(value1, value2) | 如果 value1 小于等于 value2，返回 TRUE；如果 value1 或 value2 为 NULL，返回 FALSE。 |
| value IS NULL | isNull(value) | 如果 value 为 NULL，返回 TRUE。 |
| value IS NOT NULL | isNotNull(value) | 如果 value 不为 NULL，返回 TRUE。 |
| value1 IS DISTINCT FROM value2 | isDistinctFrom(value1, value2) | 如果 value1 和 value2 是不同的值，返回 TRUE。NULL 会作为普通值参与比较，且不会返回 NULL。 |
| value1 IS NOT DISTINCT FROM value2 | isNotDistinctFrom(value1, value2) | 如果 value1 和 value2 不是不同的值，返回 TRUE。NULL 会作为普通值参与比较，且不会返回 NULL。 |
| value1 BETWEEN value2 AND value3 | betweenAsymmetric(value1, value2, value3) | 如果 value1 大于等于 value2 且小于等于 value3，返回 TRUE。 |
| value1 NOT BETWEEN value2 AND value3 | notBetweenAsymmetric(value1, value2, value3) | 如果 value1 小于 value2 或大于 value3，返回 TRUE。 |
| string1 LIKE string2 | like(string1, string2) | 如果 string1 按 Java 正则表达式和子串匹配语义匹配 string2，返回 TRUE。 |
| string1 NOT LIKE string2 | notLike(string1, string2) | 如果 string1 按 Java 正则表达式和子串匹配语义不匹配 string2，返回 TRUE。 |
| string1 LIKE string2 ESCAPE string3 | like(string1, string2, string3) | 如果 string1 匹配 SQL LIKE 模式 string2，返回 TRUE。% 匹配任意长度字符串，_ 匹配单个字符，string3 用于转义通配符。如果任一参数为 NULL，返回 NULL。 |
| string1 NOT LIKE string2 ESCAPE string3 | notLike(string1, string2, string3) | 如果 string1 不匹配 SQL LIKE 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| string1 SIMILAR TO string2 | similarTo(string1, string2) | 如果 string1 匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| string1 NOT SIMILAR TO string2 | notSimilarTo(string1, string2) | 如果 string1 不匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| string1 SIMILAR TO string2 ESCAPE string3 | similarTo(string1, string2, string3) | 如果 string1 使用 string3 作为转义字符匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| string1 NOT SIMILAR TO string2 ESCAPE string3 | notSimilarTo(string1, string2, string3) | 如果 string1 使用 string3 作为转义字符不匹配 SQL SIMILAR TO 模式 string2，返回 TRUE。如果任一参数为 NULL，返回 NULL。 |
| value1 IN (value2 [, value3]* ) | in(value1, value2 [, value3]*) | 如果 value1 存在于给定列表（value2, value3, …）中，返回 TRUE。 |
| value1 NOT IN (value2 [, value3]* ) | notIn(value1, value2 [, value3]*) | 如果 value1 不存在于给定列表（value2, value3, …）中，返回 TRUE。 |
逻辑函数
逻辑函数遵循 SQL 的三值逻辑来处理可为 NULL 的 BOOLEAN 值。AND 和 OR 会在左操作数已经决定结果时短路，不再计算右操作数。生成的 Janino 代码会根据操作数可空性使用原生运算符、条件表达式或惰性函数调用。
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| boolean1 OR boolean2 | 短路 OR 表达式 | 如果任一值为 TRUE，返回 TRUE；如果没有 TRUE 且至少一个值为 NULL，返回 NULL。 |
| boolean1 AND boolean2 | 短路 AND 表达式 | 如果任一值为 FALSE，返回 FALSE；如果没有 FALSE 且至少一个值为 NULL，返回 NULL。 |
| NOT boolean | not(boolean) | 如果 boolean 为 FALSE，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE；如果为 NULL，返回 NULL。 |
| boolean IS FALSE | isFalse(boolean) | 如果 boolean 为 FALSE，返回 TRUE；如果 boolean 为 TRUE 或 NULL，返回 FALSE。 |
| boolean IS NOT FALSE | isNotFalse(boolean) | 如果 boolean 为 TRUE 或 NULL，返回 TRUE；如果 boolean 为 FALSE，返回 FALSE。 |
| boolean IS TRUE | isTrue(boolean) | 如果 boolean 为 TRUE，返回 TRUE；如果 boolean 为 FALSE 或 NULL，返回 FALSE。 |
| boolean IS NOT TRUE | isNotTrue(boolean) | 如果 boolean 为 FALSE 或 NULL，返回 TRUE；如果 boolean 为 TRUE，返回 FALSE。 |
| boolean IS UNKNOWN | isNull(boolean) | 如果 boolean 为 NULL，返回 TRUE。 |
| boolean IS NOT UNKNOWN | isNotNull(boolean) | 如果 boolean 不为 NULL，返回 TRUE。 |
数学函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| numeric1 + numeric2 | numeric1 + numeric2 | 返回 NUMERIC1 加 NUMERIC2 的结果。 |
| numeric1 - numeric2 | numeric1 - numeric2 | 返回 NUMERIC1 减 NUMERIC2 的结果。 |
| numeric1 * numeric2 | numeric1 * numeric2 | 返回 NUMERIC1 乘 NUMERIC2 的结果。 |
| numeric1 / numeric2 | numeric1 / numeric2 | 返回 NUMERIC1 除以 NUMERIC2 的结果。 |
| numeric1 % numeric2 | numeric1 % numeric2 | 返回 numeric1 除以 numeric2 的余数（取模）。 |
| ABS(numeric) | abs(numeric) | 返回 numeric 的绝对值。 |
| CEIL(numeric) CEILING(numeric) | ceil(numeric) | 向上取整，返回大于或等于 numeric 的最小整数。 |
| FLOOR(numeric) | floor(numeric) | 向下取整，返回小于或等于 numeric 的最大整数。 |
| ROUND(numeric, int) | round(numeric) | 返回 NUMERIC 四舍五入到 INT 位小数的数字。 |
| UUID() | uuid() | 根据 RFC 4122 类型 4（伪随机生成）返回一个 UUID（通用唯一标识符）字符串（例如 "3d3c68f7-f608-473f-b60c-b0c44ad4cc4e"）。 |
字符串函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| string1 || string2 | concat(string1, string2) | 返回 STRING1 和 STRING2 连接后的字符串。 |
| CHAR_LENGTH(string) | charLength(string) | 返回 STRING 中的字符数。 |
| UPPER(string) | upper(string) | 返回大写形式的字符串。 |
| LOWER(string) | lower(string) | 返回小写形式的字符串。 |
| TRIM(string1) | trim('BOTH',string1) | 返回去除两端空格的字符串。 |
| LTRIM(string[, trimString]) | ltrim(string[, trimString]) | 返回去除开头 trimString 字符后的字符串，默认去除空格。 |
| RTRIM(string[, trimString]) | rtrim(string[, trimString]) | 返回去除末尾 trimString 字符后的字符串，默认去除空格。 |
| BTRIM(string[, trimString]) | btrim(string[, trimString]) | 返回去除开头和末尾 trimString 字符后的字符串，默认去除空格。 |
| REGEXP_REPLACE(string1, string2, string3) | regexpReplace(string1, string2, string3) | 返回将 STRING1 中所有匹配正则表达式 STRING2 的子串替换为 STRING3 后的字符串。例如，'foobar'.regexpReplace('oo|ar', '') 返回 "fb"。 |
| REGEXP_EXTRACT(string, regex[, extractIndex]) | regexpExtract(string, regex[, extractIndex]) | 返回正则表达式组 extractIndex 捕获的子串。extractIndex 默认为 0，0 表示完整匹配。输入为 NULL、无匹配、正则表达式非法或组索引非法时返回 NULL。 |
| REGEXP_EXTRACT_ALL(string, regex[, extractIndex]) | regexpExtractAll(string, regex[, extractIndex]) | 返回组 extractIndex 捕获的所有子串组成的 ARRAY<STRING>。extractIndex 默认为 1，0 表示完整匹配。无匹配时返回空数组，输入为 NULL、正则表达式非法或组索引非法时返回 NULL。 |
| REGEXP_COUNT(string, regex) | regexpCount(string, regex) | 返回正则表达式匹配的非重叠子串数量。无匹配时返回 0，输入为 NULL 或正则表达式非法时返回 NULL。 |
| REGEXP_INSTR(string, regex) | regexpInstr(string, regex) | 返回第一个匹配正则表达式的子串的 1-based 位置。无匹配时返回 0，输入为 NULL 或正则表达式非法时返回 NULL。 |
| REGEXP_SUBSTR(string, regex) | regexpSubstr(string, regex) | 返回第一个匹配正则表达式的子串。输入为 NULL、无匹配或正则表达式非法时返回 NULL。 |
| SUBSTR(string, integer1[, integer2]) | substr(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为 integer2（默认到末尾）的子串。 |
| SUBSTRING(string FROM integer1 [ FOR integer2 ]) | substring(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为 integer2（默认到末尾）的子串。 |
| OVERLAY(string1 PLACING string2 FROM integer1 [FOR integer2]) | overlay(string1, string2, integer1[, integer2]) | 从位置 integer1 开始，用 STRING2 替换 STRING1 的子串，替换长度默认为 STRING2 的长度，支持字符字符串和二进制字符串。 |
| POSITION(string1 IN string2 [ FROM integer ]) | position(string1, string2[, integer]) | 返回 STRING1 在 STRING2 中第一次出现的位置，可指定从 integer 开始查找。起始位置为 1，未找到时返回 0，支持字符字符串和二进制字符串。 |
| LOCATE(string1, string2[, integer]) | locate(string1, string2[, integer]) | 返回 STRING1 在 STRING2 中第一次出现的位置，可指定从 integer 开始查找。起始位置为 1，未找到时返回 0。 |
| INSTR(string1, string2) | instr(string1, string2) | 返回 STRING2 在 STRING1 中第一次出现的位置。起始位置为 1，未找到时返回 0。 |
| CONCAT(string1, string2,…) | concat(string1, string2,…) | 返回连接 string1、string2、… 后的字符串。例如，CONCAT('AA', 'BB', 'CC') 返回 'AABBCC'。 |
| CONCAT_WS(separator, string1, string2,...) | concatWs(separator, string1, string2,...) | 使用分隔符连接 string1、string2、... 后返回字符串。NULL 字符串参数会被跳过。 |
| LPAD(string1, integer, string2) | lpad(string1, integer, string2) | 返回使用 STRING2 左填充 STRING1 至 integer 个字符后的字符串。如果 STRING1 更长，则截断到 integer 个字符。 |
| RPAD(string1, integer, string2) | rpad(string1, integer, string2) | 返回使用 STRING2 右填充 STRING1 至 integer 个字符后的字符串。如果 STRING1 更长，则截断到 integer 个字符。 |
| REPLACE(string1, string2, string3) | replace(string1, string2, string3) | 返回将 STRING1 中所有 STRING2 替换为 STRING3 后的字符串。 |
| REPEAT(string, integer) | repeat(string, integer) | 返回将 STRING 重复 integer 次后的字符串。 |
| LEFT(string, integer) | left(string, integer) | 返回 STRING 最左侧 integer 个字符。 |
| RIGHT(string, integer) | right(string, integer) | 返回 STRING 最右侧 integer 个字符。 |
| STARTSWITH(string1, string2) | startswith(string1, string2) | 返回 STRING1 是否以 STRING2 开头，支持字符字符串和二进制字符串。 |
| ENDSWITH(string1, string2) | endswith(string1, string2) | 返回 STRING1 是否以 STRING2 结尾，支持字符字符串和二进制字符串。 |
| TO_BASE64(string | binary) | toBase64(string | binary) | 将字符或二进制字符串编码为 base64 字符串。 |
| FROM_BASE64(string) | fromBase64(string) | 将 base64 字符串按 UTF-8 解码，并假定解码后的字节是有效的 UTF-8。对于任意二进制内容，请使用 FROM_BASE64_BINARY。 |
| FROM_BASE64_BINARY(string) | fromBase64Binary(string) | 将 base64 字符串解码为 VARBINARY，并保留原始字节。 |
时间函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| LOCALTIME | localtime() | 返回本地时区的当前 SQL 时间，返回类型为 TIME(0)。 |
| LOCALTIMESTAMP | localtimestamp() | 返回本地时区的当前 SQL 时间戳，返回类型为 TIMESTAMP(3)。 |
| CURRENT_TIME | currentTime() | 返回本地时区的当前 SQL 时间，是 LOCAL_TIME 的同义词。 |
| CURRENT_DATE | currentDate() | 返回本地时区的当前 SQL 日期。 |
| CURRENT_TIMESTAMP | currentTimestamp() | 返回本地时区的当前 SQL 时间戳，返回类型为 TIMESTAMP_LTZ(3)。 |
| NOW() | now() | 返回本地时区的当前 SQL 时间戳，是 CURRENT_TIMESTAMP 的同义词。 |
| EXTRACT(timeintervalunit FROM temporal) | extract(timeintervalunit, temporal) | 从时间值中提取指定部分。支持 YEAR、QUARTER、MONTH、WEEK、DAY、DOY、DOW、HOUR、MINUTE 和 SECOND。 |
| YEAR / QUARTER / MONTH / WEEK(temporal) | extract(timeintervalunit, temporal) | 从时间值中提取年、季度、月或 ISO 周。 |
| DAYOFYEAR / DAYOFMONTH / DAYOFWEEK(temporal) | extract(timeintervalunit, temporal) | 从时间值中提取一年中的第几天、一月中的第几天或一周中的第几天。DAYOFWEEK 以星期日为 1。 |
| HOUR / MINUTE / SECOND(temporal) | extract(timeintervalunit, temporal) | 从时间值中提取小时、分钟或秒。 |
| temporal +/- INTERVAL 'value' unit | temporalPlusMonths / temporalPlusMillis | 对 DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ 值加减年 - 月或日 - 时间隔字面量。支持 INTERVAL + temporal，以及 INTERVAL '1-2' YEAR TO MONTH 和 INTERVAL '1 02:03:04' DAY TO SECOND 等复合间隔。 |
| DATE_FORMAT(timestamp, string) | dateFormat(timestamp, string) | 将时间戳转换为指定日期格式字符串的值。格式字符串与 Java 的 SimpleDateFormat 兼容。 |
| DATE_FORMAT(date, string) | dateFormat(date, string) | 将给定日期转换为指定格式字符串的值。格式字符串与 Java 的 SimpleDateFormat 兼容。 |
| DATE_FORMAT(time, string) | dateFormat(time, string) | 将给定时间转换为指定格式字符串的值。格式字符串与 Java 的 SimpleDateFormat 兼容。 |
| DATE_FORMAT_TZ(timestamp, format, timezone) | dateFormatTz(timestamp, format, timezone) | 使用给定的模式和指定的时区将时间戳或日期时间值格式化为字符串。timezone 参数可以是时区 ID（例如 'UTC'、'Asia/Shanghai'）或偏移量（如 '+08:00'）。 |
| TIMESTAMPADD(timeintervalunit, interval, timepoint) | timestampadd(timeintervalunit, interval, timepoint) | 返回 timepoint 加上 interval 后的时间戳。时间间隔的单位由第一个参数指定，应为以下值之一：SECOND、MINUTE、HOUR、DAY、MONTH 或 YEAR。 |
| TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2) | timestampDiff(timepointunit, timepoint1, timepoint2) | 返回 timepoint1 和 timepoint2 之间的（有符号）时间单位数。时间间隔的单位由第一个参数指定，应为以下值之一：SECOND、MINUTE、HOUR、DAY、MONTH 或 YEAR。 |
| TO_DATE(string1[, string2]) | toDate(string1[, string2]) | 将日期字符串 string1 按格式 string2（默认为 'yyyy-MM-dd'）转换为日期。格式参数仅适用于该字符串重载。 |
| TO_DATE(timestamp) | toDate(timestamp) | 将 TIMESTAMP、带时区的 TIMESTAMP（TIMESTAMP_TZ）或 TIMESTAMP_LTZ 值按该时间值的日历日期转换为 DATE。该重载不支持格式模式；若要解析带格式的字符串，请使用 TO_DATE(string1[, string2])。 |
| TO_TIMESTAMP(string1[, string2]) | toTimestamp(string1[, string2]) | 将日期时间字符串 string1 按格式 string2（默认为 'yyyy-MM-dd HH:mm:ss'）转换为时间戳，不带时区。 |
| TO_TIMESTAMP_LTZ(string1[, string2]) | toTimestampLtz(string1[, string2]) | 将日期时间字符串 string1 按格式 string2（默认为 'yyyy-MM-dd HH:mm:ss'）转换为时间戳，带本地时区。 |
| FROM_UNIXTIME(numeric[, string]) | fromUnixtime(NUMERIC[, STRING]) | 返回 numeric 参数的字符串格式表示（默认为 'yyyy-MM-dd HH:mm:ss'）。numeric 是表示自 '1970-01-01 00:00:00' UTC 以来秒数的内部时间戳值，例如由 UNIX_TIMESTAMP() 函数产生的值。返回值以会话时区（在 TableConfig 中指定）表示。例如，如果在 UTC 时区，FROM_UNIXTIME(44) 返回 '1970-01-01 00:00:44'，但如果在 'Asia/Tokyo' 时区则返回 '1970-01-01 09:00:44'。 |
| UNIX_TIMESTAMP() | unixTimestamp() | 获取当前 Unix 时间戳（秒）。此函数是非确定性的，意味着每条记录都会重新计算该值。 |
| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | 将日期时间字符串 string1 按格式 string2（如果未指定，默认为 yyyy-MM-dd HH:mm:ss）转换为 Unix 时间戳（秒），使用表配置中指定的时区。 如果日期时间字符串中指定了时区并以 UTC+X 格式（如 "yyyy-MM-dd HH:mm:ss.SSS X"）解析，则此函数将使用日期时间字符串中指定的时区而不是表配置中的时区。如果无法解析日期时间字符串，将返回默认值 Long.MIN_VALUE(-9223372036854775808)。 |
| DATE_ADD(date, int) | dateAdd(date, int) | 将 N 天添加到给定日期，返回格式为 'yyyy-MM-dd' 的字符串。 |
条件函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| CASE value WHEN value1_1 [, value1_2]* THEN RESULT1 (WHEN value2_1 [, value2_2 ]* THEN result_2)* (ELSE result_z) END | 嵌套三元表达式 | 当 value 第一次匹配到（valueX_1, valueX_2, …）中的值时，返回 resultX。如果没有匹配的值，如果提供了 result_z 则返回 result_z，否则返回 NULL。 |
| CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END | 嵌套三元表达式 | 当第一个 conditionX 满足时，返回 resultX。如果没有条件满足，如果提供了 result_z 则返回 result_z，否则返回 NULL。 |
| COALESCE(value1 [, value2]*) | coalesce(Object... objects) | 返回第一个不为 NULL 的参数。如果所有参数都为 NULL，则返回 NULL。返回类型是所有参数中限制最少的公共类型。如果所有参数都可为空，则返回类型也可为空。 |
| IF(condition, true_value, false_value) | condition ? true_value : false_value | 如果条件满足，返回 true_value，否则返回 false_value。例如，IF(5 > 3, 5, 3) 返回 5。 |
| IFNULL(value, replacement) | ifNull(value, replacement) | 当 value 为 NULL 时返回 replacement，否则返回 value。两个参数必须存在公共类型。仅当 replacement 可为 NULL 时，结果才可为 NULL。 |
| NULLIF(value1, value2) | nullIf(value1, value2) | 当 value1 与 value2 相等时返回 NULL，否则返回 value1。返回类型为 value1 对应的可空类型。 |
NULLIF 会对数值参数进行跨数值类型比较。例如，NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT)) 返回 NULL。该行为与当前 Transform = 运算符的类型敏感相等比较不同；对于上述混合类型比较，= 返回 FALSE。
转换函数
你可以使用 CAST( <EXPR> AS <T> ) 语法将任何有效的表达式 <EXPR> 转换为特定类型 <T>。可能的转换路径如下：
TRY_CAST( <EXPR> AS <T> ) 支持与 CAST 相同的转换路径。当某个值无法转换时返回 NULL；不支持的源类型到目标类型转换路径会在 pipeline 校验或初始化阶段被拒绝。配置错误、UDF 异常和内部程序错误不会被转换为 NULL。
| 源类型 | 目标类型 | 说明 |
|---|---|---|
| ANY | STRING | 所有类型都可以转换为 STRING。 |
| NUMERIC, STRING | BOOLEAN | 任何非零数字将被计算为 TRUE。 |
| NUMERIC | BYTE | 值必须在 Byte 范围内（-128 ~ 127）。 |
| NUMERIC | SHORT | 值必须在 Short 范围内（-32768 ~ 32767）。 |
| NUMERIC | INTEGER | 值必须在 Integer 范围内（-2147483648 ~ 2147483647）。 |
| NUMERIC | LONG | 值必须在 Long 范围内（-9223372036854775808 ~ 9223372036854775807）。 |
| NUMERIC | FLOAT | 值必须在 Float 范围内（1.40239846e-45f ~ 3.40282347e+38f）。 |
| NUMERIC | DOUBLE | 值必须在 Double 范围内（4.94065645841246544e-324 ~ 1.79769313486231570e+308）。 |
| NUMERIC | DECIMAL | 值必须在 BigDecimal(10, 0) 范围内。 |
| STRING, TIMESTAMP_TZ, TIMESTAMP_LTZ | TIMESTAMP | 字符串类型的值必须是有效的 ISO_LOCAL_DATE_TIME 格式字符串。 |
Variant 函数
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| PARSE_JSON(string[, allowDuplicateKeys]) | parseJson(string[, allowDuplicateKeys]) | 将 JSON 字符串解析为 Variant。如果 JSON 字符串无效，将抛出错误。如需返回 NULL 而不是抛出错误，请使用 TRY_PARSE_JSON 函数。 如果输入的 JSON 字符串中存在重复的键，当 allowDuplicateKeys 为 true 时，解析器将保留所有相同键的最后一个出现的字段，否则将抛出错误。allowDuplicateKeys 的默认值为 false。 |
| TRY_PARSE_JSON(string[, allowDuplicateKeys]) | tryParseJson(string[, allowDuplicateKeys]) | 如果可能，将 JSON 字符串解析为 Variant。如果 JSON 字符串无效，返回 NULL。如需抛出错误而不是返回 NULL，请使用 PARSE_JSON 函数。 如果输入的 JSON 字符串中存在重复的键，当 allowDuplicateKeys 为 true 时，解析器将保留所有相同键的最后一个出现的字段，否则将返回 NULL。allowDuplicateKeys 的默认值为 false。 |
结构函数
结构函数用于使用下标语法访问 ARRAY、MAP、ROW 和 VARIANT 类型中的元素。
| 函数 | Janino 代码 | 描述 |
|---|---|---|
| ARRAY[value, ...] | array(value, ...) | 使用给定值构造数组。所有值必须具有公共类型。 |
| MAP[key, value, ...] | map(key, value, ...) | 使用键值对构造 Map。所有键、所有值必须分别具有公共类型。 |
| ROW(value, ...) | row(value, ...) | 使用给定值构造 Row。字段名依次生成为 f0、f1 等。 |
| CARDINALITY(arrayOrMap) | cardinality(arrayOrMap) | 返回数组或 Map 的元素数量。输入为 NULL 时返回 NULL。 |
| ARRAY_CONTAINS(array, value) | arrayContains(array, value) | 返回数组是否包含给定值。 |
| ARRAY_POSITION(array, value) | arrayPosition(array, value) | 返回首个匹配值的位置，位置从 1 开始；未找到时返回 0。 |
| ELEMENT(array) | element(array) | 返回单元素数组中的元素。空数组返回 NULL；数组包含多个元素时抛出异常。 |
| array[index] | itemAccess(array, index) | 返回数组中位置 index 的元素。索引从 1 开始（SQL 标准）。如果索引超出范围或数组为 NULL，则返回 NULL。 |
| map[key] | itemAccess(map, key) | 返回 map 中与 key 关联的值。如果 key 不存在或 map 为 NULL，则返回 NULL。 |
| row[index] | itemAccess(row, index) | 返回 row 中位置 index 的字段。索引从 1 开始。索引必须是常量（不能是计算表达式），因为返回类型必须在静态阶段确定。 |
| variant[index] | itemAccess(variant, index) | 返回 variant 中位置 index 的元素（如果是数组）。索引从 1 开始（SQL 标准）。如果索引超出范围或 variant 不是数组，则返回 NULL。 |
| variant[key] | itemAccess(variant, key) | 返回 variant 中与 key 关联的值（如果是对象）。如果 key 不存在或 variant 不是对象，则返回 NULL。 |
示例
添加计算列
表达式可以用来生成新的列。例如，如果我们想基于表 web_order 在数据库 mydb 中添加两个计算列，我们可以定义一个转换规则如下：
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
引用元数据列
我们可以在投影表达式中引用元数据列。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
使用通配符来包含所有字段
通配符（*）可以用来引用表中的所有字段。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
注意：当 * 字符出现在表达式的开头时，需要添加转义反斜杠。
添加过滤条件
使用引用列时，我们可以在添加过滤规则时使用。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
transform:
- source-table: mydb.web_order
filter: id > 10 AND order_id > 100
description: filtering rows from source table
计算列可以在过滤条件中使用。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
重新设置主键
我们可以在转换规则中重新设置主键。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
transform:
- source-table: mydb.web_order
projection: id, order_id
primary-keys: order_id
description: reassign primary key example
复合主键也支持。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
注意主键列不能为空。
重新设置分区键
我们可以重新设置分区键。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
定义建表参数
额外的建表参数可以定义在转换规则中，并在创建下游表时应用。例如，给定表 web_order 在数据库 mydb 中，我们可以定义一个转换规则如下：
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
小技巧：table-options 的格式是 key1=value1,key2=value2；如果 value 中包含逗号或其他特殊字符，可以使用 table-options.delimiter 指定自定义分隔符（如 ;、|、$ 等）：
transform:
- source-table: mydb.web_order
table-options: sequence.field=gxsj,jjsj;file-index.bloom-filter.columns=jjdbh
table-options.delimiter: ";"
分类映射
在一张表同时被多个转换规则命中时， 只有第一个匹配的转换规则将应用。 举个例子，我们可以定义一个转换规则如下：
transform:
- source-table: mydb.web_order
projection: id, order_id
filter: id > 1001
- source-table: mydb.\.*
projection: \*, 'fallback' AS FALLBACK
这里，即使 mydb.web_order 表同样可以被第二条规则匹配，但因为排序靠前的第一条规则已经匹配，因此不会落入后续的 Transform 规则中。
用户自定义函数
用户自定义函数（UDF）可以在转换规则中使用。
一个满足了下面条件的类可以作为 UDF:
- 继承了
org.apache.flink.cdc.common.udf.UserDefinedFunction接口 - 有一个 public 的无参构造函数
- 有至少一个 public 方法命名为
eval
也可以：
- 重载
getReturnType方法来指示它的返回 CDC 类型 - 重载
open和close方法来执行一些初始化和清理工作
举个例子，这是一个有效的 UDF 类：
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 {
// ...
}
}
为了兼容 Flink SQL，UDF 类也可以是 Flink ScalarFunction，但有一些限制：
- 包含有参构造方法的
ScalarFunction是不支持的。 - Flink 的
DataType提示在ScalarFunction中将被忽略。 open/close生命周期钩子不会被调用。
UDF 类可以通过 user-defined-function 块进行注册：
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
注意这里的 classpath 必须是全限定名，并且对应的 jar 文件必须包含在 Flink /lib 文件夹中，或者通过 flink-cdc.sh --jar 选项传递。
UDF 配置选项
你可以通过添加 options 块来向 UDF 传递额外的配置选项。这些选项可以在 open 方法中通过 UserDefinedFunctionContext.configuration() 获取：
pipeline:
user-defined-function:
- name: query_redis
classpath: com.example.flink.cdc.udf.RedisQueryFunction
options:
hostname: localhost
port: "6379"
cache.enabled: "true"
在你的 UDF 实现中，可以通过定义 ConfigOption 实例来访问这些配置选项：
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);
// 在这里初始化你的连接...
}
public Object eval(String key) {
// 使用 hostname 和 port 查询 Redis...
}
}
options 字段是可选的。如果未指定，将会传递一个空的配置给 UDF。
在正确注册后，UDF 可以在 projection 和 filter 表达式中使用，就像内置函数一样：
transform:
- source-table: db.\.*
projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
filter: inc(id) < 100
有关 AI 模型函数和配置，请参见 AI 模型。
Python UDF
Flink CDC 支持直接在 Pipeline YAML 中定义 Python UDF。发行包内置了 flink-cdc-python
模块，通过 Pemja 在每个 TaskManager 中嵌入 Python。
transform:
- source-table: db.users
projection: ID, py_normalize(EMAIL) AS EMAIL_NORM, py_double(AGE) AS DOUBLED
pipeline:
user-defined-function:
- name: py_normalize
python-code: |
def eval(s: str) -> str:
return None if s is None else s.strip().lower()
python-executable: /usr/bin/python3
- name: py_double
python-code: |
def eval(x: int) -> int:
return None if x is None else x * 2
python-files:
- /opt/flink/python-deps
- /opt/flink/python-deps.zip
每个条目必须包含一个名为 eval 的顶层函数，并声明返回值类型。目前支持 bool、
bytes、float、int 和 str，依次映射为 Flink CDC 的 BOOLEAN、BYTES、
DOUBLE、BIGINT 和 STRING。
运行时还需满足以下条件：
- 每个 TaskManager 都要安装 Python 和
pemja==0.5.7。python-executable默认使用PATH中找到的第一个python3。 - 可选的
python-files是一个 YAML 列表，元素只能是已存在的目录或.zip压缩包； 这些路径必须在每个 TaskManager 上都可访问。压缩包解压后才会加入 Python import path。 python-code不能与classpath或options同时使用。每个 Python 函数应单独声明为 一个 UDF 条目。
Embedding AI 模型
内置 AI 模型可以在 transform 规则中使用。
为了使用内置 AI 模型，你需要下载内置模型的 jar，然后在 flink-cdc.sh 命令中添加 --jar {$BUILT_IN_MODEL_PATH}。
如何定义一个 Embedding AI 模型：
pipeline:
model:
- model-name: CHAT
class-name: OpenAIChatModel
openai.model: gpt-4o-mini
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
注意：
model-name是一个通用的必填参数，用于所有支持的模型，表示在projection或filter中调用的函数名称。class-name是一个通用的必填参数，用于所有支持的模型，可用值可以在所有支持的模型中找到。openai.model，openai.host，openai.apiKey和openai.chat.prompt是在各个模型中特别的可选参数。
如何使用一个 Embedding AI 模型：
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
这里，GET_EMBEDDING 是通过 model-name 在 pipeline 中定义的。
所有支持的模型
下面列出了所有支持的模型：
OpenAIChatModel
| 参数 | 类型 | 是否必填 | 含义 |
|---|---|---|---|
| openai.model | STRING | 必填 | 要调用的模型名称，例如："gpt-4o-mini"，可用选项有 "gpt-4o-mini"、"gpt-4o"、"gpt-4-32k"、"gpt-3.5-turbo"。 |
| openai.host | STRING | 必填 | 要连接的模型服务器地址，例如：http://langchain4j.dev/demo/openai/v1。 |
| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。 |
| openai.chat.prompt | STRING | 可选 | 与 OpenAI 聊天的提示词，例如："Please summary this"。 |
OpenAIEmbeddingModel
| 参数 | 类型 | 是否必填 | 含义 |
|---|---|---|---|
| openai.model | STRING | 必填 | 要调用的模型名称，例如："text-embedding-3-small"，可用选项有 "text-embedding-3-small"、"text-embedding-3-large"、"text-embedding-ada-002"。 |
| openai.host | STRING | 必填 | 要连接的模型服务器地址，例如：http://langchain4j.dev/demo/openai/v1。 |
| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。 |
Copyright © 2026 The Apache Software Foundation. Apache Flink, Flink, and the Flink logo are trademarks of The Apache Software Foundation.