Flink CDC自定义函数

Java UDF 定义

如果您希望使用 Java 语言编写自定义函数，您需要首先创建一个 Maven 项目，并在pom.xml中引入基础依赖包：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-cdc-common</artifactId>
    <version>${CDC 社区版本号}</version>
    <scope>provided</scope>
</dependency>

请参考下表选定使用的社区版本号：

满足以下要求的Java类可以作为Flink CDC数据摄入作业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 {
        // ...
    }
}

Java UDF 注册

通过在Flink CDC数据摄入作业的pipeline块中加入如下所示的定义即可注册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

Python UDF（公测中）

您可以直接在 Flink CDC 数据摄入 YAML 作业中使用 Python 语言编写内联自定义函数，无需创建 Maven 项目或额外编译 JAR 包。

def eval(value: str) -> str:
    return value.strip().lower()

import re

EMAIL_PATTERN = re.compile(r"\s+")

def normalize(value):
    return EMAIL_PATTERN.sub("", value).lower()

def eval(value: str) -> str:
    if value is None: return None
    return normalize(value)

pipeline:
  user-defined-function:
    - name: normalize_email
      python-code: |
        def eval(value: str) -> str:
            if value is None:
                return None
            return value.strip().lower()

    - name: double_age
      python-code: |
        def eval(value: int) -> int:
            if value is None:
                return None
            return value * 2

UDF 使用

在完成 UDF 函数注册后，即可在 Transform 块中，像内置函数一样直接调用 UDF 函数。代码示例如下：

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

Flink SQL UDF兼容性

继承自ScalarFunction的Flink UDF函数也可以直接作为CDC YAML UDF函数注册并使用，但存在以下限制：

• 不支持带参数的ScalarFunction。

• Flink风格的TypeInformation类型标注会被忽略。

• open和close生命周期钩子函数不会被调用。