服装网站建设策划案,谷歌搜索引擎下载,十大网络舆情案例,东莞网站建设信科分公司Flink CDC系列之#xff1a;学习理解核心概念——Transform Transform参数元数据字段函数比较函数逻辑函数字符串函数时间函数条件函数 示例添加计算列参考元数据列使用通配符投影所有字段添加过滤规则重新分配主键重新分配分区键指定表创建配置分类映射用户定义函数已知限制 … Flink CDC系列之学习理解核心概念——Transform Transform参数元数据字段函数比较函数逻辑函数字符串函数时间函数条件函数 示例添加计算列参考元数据列使用通配符投影所有字段添加过滤规则重新分配主键重新分配分区键指定表创建配置分类映射用户定义函数已知限制 Transform
Transform模块帮助用户根据表中的数据列进行数据列的删除和扩展。 此外它还可以帮助用户在同步过程中过滤一些不必要的数据。
参数
为了描述转换规则可以使用以下参数
参数含义可选/必需source-table源表id支持正则表达式必需projection投影规则支持类似SQL中select子句的语法可选filter过滤规则支持类似SQL中where子句的语法可选primary-keysSink 表主键以逗号分隔可选partition-keys接收表分区键以逗号分隔可选table-options用于自动创建表时配置表创建语句可选description变换规则说明可选
可以在一个管道 YAML 文件中声明多个规则。
元数据字段
字段定义
有一些隐藏列用于访问元数据信息。它们仅在转换规则中明确引用时才会生效。
字段数据类型描述namespace_nameString包含该行的命名空间的名称。schema_nameString包含该行的架构的名称。table_nameString包含该行的表的名称。data_event_typeString数据改变事件的操作类型。
元数据关系 函数
Flink CDC 使用 Calcite 解析表达式并使用 Janino 脚本通过函数调用来评估表达式。
比较函数
FunctionJanino Code描述value1 value2valueEquals(value1, value2)如果 value1 等于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value1 value2!valueEquals(value1, value2)如果 value1 不等于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value1 value2value1 value2如果 value1 大于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value1 value2value1 value2如果 value1 大于或等于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value1 value2value1 value2如果 value1 小于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value1 value2value1 value2如果 value1 小于或等于 value2则返回 TRUE如果 value1 或 value2 为 NULL则返回 FALSE。value IS NULLnull value如果值为 NULL则返回 TRUE。value IS NOT NULLnull ! value如果值不为 NULL则返回 TRUE。value1 BETWEEN value2 AND value3betweenAsymmetric(value1, value2, value3)如果 value1 大于或等于 value2 且小于或等于 value3则返回 TRUE。value1 NOT BETWEEN value2 AND value3notBetweenAsymmetric(value1, value2, value3)如果 value1 小于 value2 或大于 value3则返回 TRUE。string1 LIKE string2like(string1, string2)如果 string1 与模式 string2 匹配则返回 TRUE。string1 NOT LIKE string2notLike(string1, string2)如果 string1 与模式 string2 不匹配则返回 TRUE。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。
逻辑函数
FunctionJanino Code描述boolean1 OR boolean2boolean1boolean1 AND boolean2boolean1 boolean2如果 BOOLEAN1 和 BOOLEAN2 都为 TRUE则返回 TRUE。NOT boolean!boolean如果布尔值为 FALSE则返回 TRUE如果布尔值为 TRUE则返回 FALSE。boolean IS FALSEfalse boolean如果布尔值为 FALSE则返回 TRUE如果布尔值为 TRUE则返回 FALSE。boolean IS NOT FALSEtrue boolean如果 BOOLEAN 为 TRUE则返回 TRUE如果 BOOLEAN 为 FALSE则返回 FALSE。boolean IS TRUEtrue boolean如果 BOOLEAN 为 TRUE则返回 TRUE如果 BOOLEAN 为 FALSE则返回 FALSE。boolean IS NOT TRUEfalse boolean如果布尔值为 FALSE则返回 TRUE如果布尔值为 TRUE则返回 FALSE。
字符串函数
FunctionJanino Code描述string1string2CHAR_LENGTH(string)charLength(string)返回 STRING 中的字符数。UPPER(string)upper(string)返回大写的字符串。LOWER(string)lower(string)返回小写的字符串。TRIM(string1)trim(‘BOTH’,string1)返回删除两侧空格的字符串。REGEXP_REPLACE(string1, string2, string3)regexpReplace(string1, string2, string3)返回 STRING1 中的字符串其中所有与正则表达式 STRING2 匹配的子字符串均被 STRING3 连续替换。例如‘foobar’.regexpReplace(‘ooSUBSTRING(string FROM integer1 [ FOR integer2 ])substring(string,integer1,integer2)返回从位置 INT1 开始、长度为 INT2默认到末尾的 STRING 子字符串。CONCAT(string1, string2,…)concat(string1, string2,…)返回连接 string1、string2、… 的字符串。例如CONCAT(‘AA’, ‘BB’, ‘CC’) 返回 ‘AABBCC’。
时间函数
FunctionJanino Code描述LOCALTIMElocaltime()返回本地时区的当前SQL时间返回类型为TIME(0)。LOCALTIMESTAMPlocaltimestamp()返回当前SQL本地时区的时间戳返回类型为TIMESTAMP(3)。CURRENT_TIMEcurrentTime()返回本地时区的当前 SQL 时间这是 LOCAL_TIME 的同义词。CURRENT_DATEcurrentDate()返回本地时区的当前 SQL 日期。CURRENT_TIMESTAMPcurrentTimestamp()返回当前SQL本地时区的时间戳返回类型为TIMESTAMP_LTZ(3)。NOW()now()返回本地时区的当前 SQL 时间戳这是 CURRENT_TIMESTAMP 的同义词。DATE_FORMAT(timestamp, string)dateFormat(timestamp, string)将时间戳转换为日期格式字符串指定格式的字符串值。格式字符串与 Java 的 SimpleDateFormat 兼容。TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2)timestampDiff(timepointunit, timepoint1, timepoint2)返回时间点 1 和时间点 2 之间的时间点单位的有符号数。间隔的单位由第一个参数指定该参数应为以下值之一SECOND、MINUTE、HOUR、DAY、MONTH 或 YEAR。TO_DATE(string1[, string2])toDate(string1[, string2])将格式为 string2 的日期字符串 string1默认为“yyyy-MM-dd”转换为日期。TO_TIMESTAMP(string1[, string2])toTimestamp(string1[, string2])将格式为 string2 的日期时间字符串 string1默认情况下为“yyyy-MM-dd HH:mm:ss”转换为不带时区的时间戳。
条件函数
FunctionJanino Code描述CASE value WHEN value1_1 [, value1_2]* THEN RESULT1 (WHEN value2_1 [, value2_2 ]* THEN result_2)* (ELSE result_z) END嵌套三元表达式当值第一次包含在值 X_1、值 X_2、…中时返回 resultX。当没有值匹配时如果提供了 result_z则返回 result_z否则返回 NULL。CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END嵌套三元表达式第一个条件满足时返回resultX不满足条件时若有条件则返回结果否则返回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。
示例
添加计算列
求值表达式可用于生成新列。例如如果我们想基于数据库 mydb 中的表 web_order 附加两个计算列我们可以定义一个转换规则如下所示
transform:- source-table: mydb.web_orderprojection: id, order_id, UPPER(product_name) as product_name, localtimestamp as new_timestampdescription: append calculated columns based on source table参考元数据列
我们可以在投影表达式中引用元数据列。例如给定数据库 mydb 中的表 web_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: id, order_id, __namespace_name__ || . || __schema_name__ || . || __table_name__ identifier_namedescription: access metadata columns from source table使用通配符投影所有字段
通配符 (*) 可用于引用表中的所有字段。例如给定数据库 mydb 中的两个表 web_order 和 app_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: \*, UPPER(product_name) as product_namedescription: project fields with wildcard character from source table- source-table: mydb.app_orderprojection: UPPER(product_name) as product_name, *description: project fields with wildcard character from source table注意当表达式开头出现 * 字符时需要使用转义反斜杠。
添加过滤规则
使用引用列在数据库mydb中的表web_order中添加过滤规则时我们可以定义一个转换规则如下
transform:- source-table: mydb.web_orderfilter: id 10 AND order_id 100description: filtering rows from source table计算列也可以用于过滤条件。例如给定数据库 mydb 中的表 web_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: id, order_id, UPPER(province) as new_province filter: new_province SHANGHAIdescription: filtering rows based on computed columns重新分配主键
我们可以在转换规则中重新分配主键。例如给定数据库 mydb 中的表 web_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: id, order_idprimary-keys: order_iddescription: reassign primary key example还支持复合主键
transform:- source-table: mydb.web_orderprojection: id, order_id, UPPER(product_name) as product_nameprimary-keys: order_id, product_namedescription: reassign composite primary keys example重新分配分区键
我们可以在转换规则中重新分配分区键。例如给定数据库 mydb 中的表 web_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: id, order_id, UPPER(product_name) as product_namepartition-keys: product_namedescription: reassign partition key example指定表创建配置
可以在转换规则中定义额外选项这些选项将在创建下游表时应用。给定数据库 mydb 中的表 web_order我们可以定义转换规则如下
transform:- source-table: mydb.web_orderprojection: id, order_id, UPPER(product_name) as product_nametable-options: commentweb orderdescription: auto creating table options example提示table-options的格式为key1value1,key2value2。
分类映射
可以定义多个转换规则来对输入数据行进行分类并应用不同的处理。只有第一个匹配的转换规则才会应用。例如我们可以定义如下转换规则
transform:- source-table: mydb.web_orderprojection: id, order_idfilter: UPPER(province) SHANGHAIdescription: classification mapping example- source-table: mydb.web_orderprojection: order_id as id, id as order_idfilter: UPPER(province) BEIJINGdescription: classification mapping example用户定义函数
用户定义函数 (UDF) 可用于转换规则。
如果满足以下条件类可用作 UDF
实现 org.apache.flink.cdc.common.udf.UserDefinedFunction 接口具有无参数的公共构造函数至少有一个名为 eval 的公共方法
它还可以
覆盖 getReturnType 方法以指示其返回 CDC 类型覆盖 open 和 close 方法以执行一些初始化和清理工作
例如这是一个有效的 UDF 类
public class AddOneFunctionClass implements UserDefinedFunction {public Object eval(Integer num) {return num 1;}Overridepublic DataType getReturnType() {return DataTypes.INT();}Overridepublic void open() throws Exception {// ...}Overridepublic void close() throws Exception {// ...}
}为了简化从 Flink SQL 到 Flink CDC 的迁移Flink ScalarFunction 也可以用作转换 UDF但有一些限制
不支持具有带参数的构造函数的 ScalarFunction。ScalarFunction 中的 Flink 样式类型提示将被忽略。不会调用打开/关闭生命周期钩子。
可以通过添加用户定义函数块来注册 UDF 类
pipeline:user-defined-function:- name: addoneclasspath: org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass- name: formatclasspath: org.apache.flink.cdc.udf.examples.java.FormatFunctionClass请注意给定的类路径必须是完全限定的并且相应的 jar 文件必须包含在 Flink /lib 文件夹中或者使用 flink-cdc.sh --jar 选项传递。
正确注册后UDF 可以在投影和过滤表达式中使用就像内置函数一样
transform:- source-table: db.\.*projection: *, inc(inc(inc(id))) as inc_id, format(id, id - %d) as formatted_idfilter: inc(id) 100已知限制
目前转换不适用于路由规则。它将在未来版本中得到支持。计算列不能引用最终投影结果中不存在的修剪列。这将在未来版本中修复。不支持具有不同架构的表的常规匹配。如有必要需要编写多个规则。
