本文中含有需要您注意的重要提示信息，忽略该信息可能对您的业务造成影响，请务必仔细阅读。
本文为您介绍数据摄入有关的source、sink、transform、route和pipeline模块的开发参考。
支持的连接器
连接器 | 支持类型 | |
Source | Sink | |
× | √ | |
× | √ | |
说明 支持连接RDS MySQL版、PolarDB MySQL版及自建MySQL。 | √ | × |
× | √ | |
× | √ | |
× | √ | |
× | √ | |
source模块
source模块定义数据摄入的数据源端，目前仅支持MySQL作为数据源。
语法结构
source:
type: mysql
name: mysql source
xxx: ...具体配置请查看对应连接器的数据摄入部分。
sink模块
sink模块定义数据摄入的目标端，目前支持的系统包括消息队列Kafka、Upsert Kafka、实时数仓Hologres、流式数据湖仓Paimon、StarRocks和Print。
语法结构
sink:
type: hologres
name: hologres sink
xxx: ...具体配置请查看对应连接器的数据摄入部分。
transform模块
您可以在YAML作业的transform语句块中填写若干规则信息，从而实现源表中数据的投影、计算和过滤等功能。
语法结构
transform:
- source-table: db.tbl1
projection: ...
filter: ...
- source-table: db.tbl2
projection: ...
filter: ...配置项
参数 | 含义 | 是否必填 | 备注 |
| 指定生效上游表。 | 是 | 支持使用正则表达式。 |
| 指定用于保留部分上游列的投影规则。 | 否 | 使用的句法与SQL SELECT语句类似。 不填则不追加或删除任何列。 |
| 行过滤规则。 | 否 | 使用的句法与SQL WHERE语句类似。 不填则不过滤任何行。 |
| 设定transform后Schema的主键列表。 | 否 | 不填则保留原Schema的主键定义。主键列表使用英文逗号（ |
| 设定transform的分区键列表。 | 否 | 不填则保留原Schema的分区键定义，分区键列表使用英文逗号（ |
| 需要传递给Sink的额外配置信息。 | 否 | Options选项，例如Paimon Sink的分桶数、注释等信息。 |
| 该transform块的描述信息。 | 否 | 无。 |
计算列
您可以在projection规则中使用<Expression> AS <ColName>句法来定义计算列，表达式将对上游的每条数据分别求值后填入相应列。
计算列的表达式不可以引用其他计算列的值，即使被引用的列出现在该计算列之前。例如a, b AS c, c AS d不是合法的projection表达式。
例如，在接收到来自上游db.tbl表的[+I, id = 1]数据记录时，将其转化为[+I, id = 1, inc_id = 2]数据行并发送给下游。
transform:
- source-table: db.tbl
projection: id, id + 1 AS inc_id通配符
如果您希望将源表中的所有列以及后续追加的新列按原样发送给下游，则可以在projection规则中使用星号（*）通配符。
如果一个projection规则中没有使用通配符（*），则其产生的Schema就是固定的，并且始终与projection规则中写出的版本保持一致。
例如，*, 'extras' AS extras表示会在上游Schema的列尾追加额外的列，并持续将上游的表结构变更发送给下游。
transform:
- source-table: db.tbl
projection: \*, 'extras' AS extras元数据列
在编写projection规则时，可以将以下预先定义的元数据列作为普通数据列使用：
请勿定义与元数据列同名的普通数据列。
元数据列名称 | 数据类型 | 说明 |
| String | 这条数据变更记录对应源表的Namespace名称。 |
| String | 这条数据变更记录对应源表的Schema名称。 |
| String | 这条数据变更记录对应源表的Table名称。 |
| String | 这条数据变更记录对应的操作类型（ 重要 由于CDC Event总是将一次更新对应的Update Before和Update After打包为一条事件，因此 |
例如，将上游表的全限定名称写入计算列中，并发送给下游。
transform:
- source-table: \.*.\.*
projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier各个数据库连接器对Namespace、Schema和Table名称的映射关系如下表所示。
数据库类型 | Namespace名称 | Schema名称 | 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 |
注意事项
修改transform模块的语句后，不能从已有的状态恢复，需要进行无状态启动。
通常情况下，projection和filter语句无需使用引号包裹。
transform: - projection: a, b, c # 等价于 - projection: "a, b, c"然而，如果Projection表达式的第一个字符为
*、'等特殊字符，则整行表达式可能无法被作为合法的YAML字符串字面量解析。此时需要手动使用引号包裹整个表达式，或是使用\转义：transform: - projection: *, 42 # 不是合法的YAML - projection: '*, 42' # OK - projection: \*, 42 # OK
route模块
您可以在YAML作业的route模块中定义包含若干条route规则的语句块，描述上游表到下游表的复杂拓扑结构。
语法结构
route:
- source-table: db.tbl1
sink-table: sinkdb.tbl1
- source-table: db.tbl2
sink-table: sinkdb.tbl2配置项
参数 | 含义 | 是否必填 | 备注 |
| 指定生效上游表。 | 是 | 支持使用正则表达式。 |
| 指定数据路由的目标位置。 | 是 | 无。 |
| 在使用模式匹配功能时，用于指代上游表名的字符串。 | 否 | 例如，当replace-symbol设置为 |
| 该route块的描述信息。 | 否 | 无。 |
使用方法
一对一路由
将上游表mydb.web_order中的数据路由到下游表mydb.ods_web_order。
route:
- source-table: mydb.web_order
sink-table: mydb.ods_web_order
description: sync table to one destination table with given prefix ods_合并分库分表
将上游mydb数据库中的所有表合并到下游mydb.merged表中。
route:
- source-table: mydb.\.*
sink-table: mydb.merged
description: sync sharding tables to one destination table多路由规则
可以在一个route块中使用YAML列表符号（-）定义多条规则，它们会同时生效。
route:
- source-table: mydb.orders
sink-table: ods_db.ods_orders
description: sync orders table to orders
- source-table: mydb.shipments
sink-table: ods_db.ods_shipments
description: sync shipments table to ods_shipments
- source-table: mydb.products
sink-table: ods_db.ods_products
description: sync products table to ods_products模式匹配
将source_db数据库中的全部表一一对应地同步到sink_db中，并保持表名不变。
route:
- source-table: source_db.\.*
sink-table: sink_db.<>
replace-symbol: <>
description: route all tables in source_db to sink_db使用replace-symbol参数定义的<>特殊字符串会被表名替代，从而实现源表到汇表的一一对应。
数据分发
将同一张表的数据分发给多个下游表，只需定义多条路由规则即可。例如将mydb.orders的数据会被同时分发到sink_db和backup_sink_db两个数据库中。
route:
- source-table: mydb.orders
sink-table: sink_db.orders
- source-table: mydb.orders
sink-table: backup_sink_db.orders注意事项
修改route模块的语句后，不能从已有的状态恢复，需要进行无状态启动。
pipeline模块
您可以在pipeline模块配置数据摄入YAML作业的整体配置。
语法结构
pipeline:
name: CDC YAML job
schema.change.behavior: lenient配置项
参数 | 说明 | 是否必填 | 数据类型 | 默认值 | 备注 |
name | 数据摄入YAML名称。 | 否 | STRING | Flink CDC Pipeline Job | 无。 |
schema.change.behavior | Schema变更行为配置。 | 否 | STRING | lenient | 可配置的值如下，详见Schema变更行为配置。
|
Schema变更行为配置
数据摄入YAML作业支持将数据源的Schema变更同步到下游目标端，例如创建表、添加列、重命名列、更改列类型、删除列和删除表等。下游目标端可能不支持全部的Schema变更，您可以通过schema.change.behavior配置来修改Schema变更发生时目标端的处理方式。
Schema变更模式
模式 | 说明 |
lenient（默认） | 数据摄入YAML作业会对Schema变更进行转换成目标端可处理的变更并发送，遵循以下规则：
|
exception | 不允许任何Schema变更行为。 当目标端不支持处理Schema变更时，可以使用此模式。收到Schema变更事件时，数据摄入YAML作业会抛出异常。 |
evolve | 数据摄入YAML作业会将所有Schema更改应用于目标端。 如果Schema变更在目标端应用失败，数据摄入YAML作业会抛出异常并触发故障重启。 |
try_evolve | 数据摄入YAML作业会尝试将Schema变更应用到目标端，如果目标端不支持处理发送的Schema变更，数据摄入YAML作业不会失败重启，尝试通过转换后续数据方式进行处理。 警告 try_evolve模式下，如果发生Schema变更应用失败，可能导致上游后续到来的数据出现部分列丢失、被截断等情况。 |
ignore | 所有Schema变更都不会应用于目标端。 当您的目标端尚未准备好进行任何Schema变更，想要继续从未更改的列中接收数据时，可以使用此模式。 |
控制目标端接收的Schema变更
在某些场景下，不需要所有Schema变更同步到目标端。例如，允许新增列但禁止删除列来避免删除已有的数据。
您可以通过在sink模块中设置include.schema.changes和exclude.schema.changes选项来控制。
参数 | 说明 | 是否必填 | 数据类型 | 默认值 | 备注 |
include.schema.changes | 支持应用的Schema变更。 | 否 | List<String> | 无 | 默认支持所有变更。 |
exclude.schema.changes | 不支持应用的Schema变更。 | 否 | List<String> | 无 | 优先级高于 |
以下是可配置架构变更事件类型的完整列表：
事件类型 | 说明 |
| 新增列。 |
| 变更列类型。 |
| 创建表。 |
| 删除列。 |
| 删除表。 |
| 修改列名。 |
| 清空数据。 |
Schema变更支持部分匹配。例如，传入drop相当于同时传入drop.column和 drop.table。
代码示例
示例1：Schema变更行为配置为evolve
source:
type: mysql
name: MySQL Source
hostname: ${mysql.hostname}
port: ${mysql.port}
username: ${mysql.username}
password: ${mysql.password}
tables: ${mysql.source.table}
server-id: 7601-7604
sink:
type: values
name: Values Sink
print.enabled: true
sink.print.logger: true
pipeline:
name: mysql to print job
schema.change.pipeline: evolve示例2：支持创建表和列相关事件，不支持删除列
source:
type: mysql
name: MySQL Source
hostname: ${mysql.hostname}
port: ${mysql.port}
username: ${mysql.username}
password: ${mysql.password}
tables: ${mysql.source.table}
server-id: 7601-7604
sink:
type: values
name: Values Sink
print.enabled: true
sink.print.logger: true
include.schema.changes: [create.table, column] # 匹配了 CreateTable、AddColumn、AlterColumnType、RenameColumn、和 DropColumn 事件
exclude.schema.changes: [drop.column] # 排除了 DropColumn 事件
pipeline:
name: mysql to print job
schema.change.pipeline: evolve函数
内置函数
CDC YAML提供了丰富的内置函数，可以直接在transform模块中的projection和filter表达式中使用。
比较函数
除非特别说明，否则以下内置函数在输入参数包含NULL时均返回NULL。
函数 | 说明 |
value1 = value2 | 如果value1等于value2，则返回TRUE；否则返回FALSE。 |
value1 <> value2 | 如果value1不等于value2，则返回TRUE；否则返回FALSE。 |
value1 > value2 | 如果value1大于value2，则返回TRUE；否则返回FALSE。 |
value1 >= value2 | 如果value1大于或等于value2，则返回TRUE；否则返回FALSE。 |
value1 < value2 | 如果value1小于value2，则返回TRUE；否则返回FALSE。 |
value1 <= value2 | 如果value1小于或等于value2，则返回TRUE；否则返回FALSE。 |
value IS NULL | 如果value是NULL，则返回TRUE；否则返回FALSE。 |
value IS NOT NULL | 如果value不是NULL，则返回TRUE；否则返回FALSE。 |
value1 BETWEEN value2 AND value3 | 如果value1的值介于value2和value3之间，则返回TRUE；否则返回FALSE。 |
value1 NOT BETWEEN value2 AND value3 | 如果value1的值并非介于value2和value3之间，则返回TRUE；否则返回FALSE。 |
string1 LIKE string2 | 如果string1的值与string2定义的模式匹配，则返回TRUE；否则返回FALSE。 |
string1 NOT LIKE string2 | 如果string1的值与string2定义的模式不匹配，则返回TRUE；否则返回FALSE。 |
value1 IN (value2 [, value3]* ) | 如果value1的值存在于[value2, value3, ...]列表中，则返回TRUE；否则返回FALSE。 |
value1 NOT IN (value2 [, value3]* ) | 如果value1的值不存在于[value2, value3, ...]列表中，则返回TRUE；否则返回FALSE。 |
逻辑函数
函数 | 说明 |
boolean1 OR boolean2 | 如果boolean1和boolean2至少有一个为TRUE，则返回TRUE。 |
boolean1 AND boolean2 | 如果boolean1和boolean2均为TRUE，则返回TRUE。 |
NOT boolean | 如果boolean为TRUE，则返回FALSE；如果boolean是FALSE，则返回TRUE。 |
boolean IS FALSE | 如果boolean为TRUE，则返回FALSE；如果boolean是FALSE，则返回TRUE。 |
boolean IS NOT FALSE | 如果boolean为TRUE，则返回TRUE；如果boolean是FALSE，则返回FALSE。 |
boolean IS TRUE | 如果boolean为TRUE，则返回TRUE；如果boolean是FALSE，则返回FALSE。 |
boolean IS NOT TRUE | 如果boolean为TRUE，则返回FALSE；如果boolean是FALSE，则返回TRUE。 |
算数函数
函数 | 说明 |
numeric1 + numeric2 | 返回numeric1加上numeric2的值。 |
numeric1 - numeric2 | 返回numeric1减去numeric2的值。 |
numeric1 * numeric2 | 返回numeric1乘以numeric2的值。 |
numeric1 / numeric2 | 返回numeric1除以numeric2的值。 |
numeric1 % numeric2 | 返回numeric1对numeric2取模的值。 |
ABS(numeric) | 返回numeric的绝对值。 |
CEIL(numeric) | 返回numeric向上取整的值。 |
FLOOR(numeric) | 返回numeric向下取整的值。 |
ROUND(numeric, int) | 返回numeric四舍五入到小数点后n位的值。 |
UUID() | 生成一个全局唯一ID（UUID）字符串（例如“3d3c68f7-f608-473f-b60c-b0c44ad4cc4e”）。 使用RFC 4122 type 4方法伪随机生成。 |
字符串函数
函数 | 说明 |
string1 || string2 | 返回string1和string2拼接而成的字符串。 重要 请勿将其与逻辑或运算符混淆。 |
CHAR_LENGTH(string) | 返回string字符串中的字符数。 |
UPPER(string) | 返回string的大写形式字符串。 |
LOWER(string) | 返回string的小写形式字符串。 |
TRIM(string1) | 删除string两侧的空白字符。 |
REGEXP_REPLACE(string1, string2, string3) | 将string1中所有满足string2模式的子串替换为string3。 例如， |
SUBSTRING(string FROM integer1 [ FOR integer2 ]) | 返回string从第integer1到第integer2个字符的子串。 说明 在不提供 |
CONCAT(string1, string2,…) | 返回将string1、string2、…拼接在一起形成的新字符串。 例如， |
时间函数
函数 | 说明 |
LOCALTIME | 返回当前时区下的本地时间，返回类型为 |
LOCALTIMESTAMP | 返回当前时区下的本地时间戳，返回类型为 |
CURRENT_TIME | 返回当前时区下的本地时间，与LOCAL_TIME相同。 |
CURRENT_DATE | 返回当前时区下的本地日期。 |
CURRENT_TIMESTAMP | 返回当前时区下的本地时间戳。返回类型为 |
NOW() | 返回当前时区下的本地时间戳，与CURRENT_TIMESTAMP相同。 |
DATE_FORMAT(timestamp, string) | 将传入的时间戳按指定的格式化字符串string进行格式化。 说明 格式化字符串与Java中的SimpleDateFormat格式兼容。 |
TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2) | 计算timepoint1和timepoint2之间差距多少timepointunit单位。 timepointunit可被指定为SECOND、MINUTE、HOUR、DAY、MONTH或YEAR。 |
TO_DATE(string1[, string2]) | 将传入的日期字符串string1按string2指定的格式转化为DATE类型。 说明 在不指定格式化字符串string2时，默认采用 |
TO_TIMESTAMP(string1[, string2]) | 将传入的时间戳字符串string1按string2指定的格式转化为不带时区信息的TIMESTAMP类型。 说明 在不指定格式化字符串string2时，默认采用 |
在进行projection和filter表达式求值时，可以保证其中每个子表达式所得到的时间点都一致。例如，NOW() AS t1, NOW() AS t2, NOW() AS t3得到的t1、t2、t3一定对应同一个时间戳，无论其求值时间和顺序如何。
条件函数
函数 | 说明 |
CASE value WHEN value1_1 [, value1_2]* THEN RESULT1 (WHEN value2_1 [, value2_2 ]* THEN result_2)* (ELSE result_z) END | 依次检查value值是否等于WHEN子句给出的值，并返回第一个相等子句的RESULT值。 如果没有任何子句满足条件，则返回ELSE子句指定的值。如果没有指定ELSE子句，则返回NULL。 |
CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END | 依次检查value值是否满足每个WHEN子句给出的条件，并返回第一个满足条件子句的RESULT值。 如果没有任何子句满足条件，则返回ELSE子句指定的值。如果没有指定ELSE子句，则返回NULL。 |
COALESCE(value1 [, value2]*) | 返回[value1、value2、……]列表中第一个不为NULL的元素。如果列表中所有元素均为NULL，则返回NULL。 |
IF(condition, true_value, false_value) | 如果condition子句对应的条件为真，则返回true_value；否则返回false_value。 |
UDF函数
CDC YAML也支持使用Java语言编写自定义UDF函数，并像内置函数一样调用。
UDF函数定义
满足以下要求的Java类可以作为CDC YAML UDF函数使用：
实现了
org.apache.flink.cdc.common.udf.UserDefinedFunction接口。拥有一个公共无参构造器。
至少含有一个名为
eval的公共方法。
UDF函数类可以通过@Override以下接口来实现更精细的语义控制：
重写
getReturnType方法来手动指定方法的返回类型。重写
open和close方法来插入生命周期函数。
例如，将传入的整型参数增加1后返回的UDF函数定义如下。
public class AddOneFunctionClass implements UserDefinedFunction {
public Object eval(Integer num) {
return num + 1;
}
@Override
public DataType getReturnType() {
// 由于eval函数的返回类型不明确，需要
// 使用getReturnType写明确指定类型
return DataTypes.INT();
}
@Override
public void open() throws Exception {
// ...
}
@Override
public void close() throws Exception {
// ...
}
}UDF函数注册
通过在CDC YAMLpipeline块中加入如下所示的定义即可注册UDF函数：
pipeline:
user-defined-function:
- name: inc
classpath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass
- name: format
classpath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass此处类路径对应的JAR包需要作为外部依赖上传。
UDF函数名称可以在此处任意调整，无需与UDF类名一致。
UDF函数使用
在完成UDF函数注册后，即可在projection和filter语句块中，像内置函数一样直接调用UDF函数。代码示例如下。
transform:
- source-table: db.\.*
projection: "*, inc(inc(inc(id))) as inc_id, format(id, 'id -> %d') as formatted_id"
filter: inc(id) < 100Flink ScalarFucntion兼容性
继承自ScalarFunction的Flink UDF函数也可以直接作为CDC YAML UDF函数注册并使用，但存在以下限制：
不支持带参数的
ScalarFunction。Flink风格的TypeInformation类型标注会被忽略。
open和close生命周期钩子函数不会被调用。
相关文档
数据摄入YAML作业开发的操作步骤，详情请参见数据摄入YAML作业开发（公测中）。