SQL用户定义函数

SQL 转换插件的用户定义函数 (UDF)

描述

使用UDF SPI扩展SQL转换函数库。

UDF API

package org.apache.seatunnel.transform.sql.zeta;

public interface ZetaUDF {
    /**
     * Function name
     *
     * @return function name
     */
    String functionName();

    /**
     * The type of function result
     *
     * @param argsType input arguments type
     * @return result type
     */
    SeaTunnelDataType<?> resultType(List<SeaTunnelDataType<?>> argsType);

    /**
     * Evaluate
     *
     * @param args input arguments
     * @return result value
     */
    Object evaluate(List<Object> args);
}

UDF 实现示例

将这些依赖项添加到您的 Maven 项目，并使用 provided 作用域。

<dependencies>
    <dependency>
        <groupId>org.apache.seatunnel</groupId>
        <artifactId>seatunnel-transforms-v2</artifactId>
        <version>2.3.2</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.seatunnel</groupId>
        <artifactId>seatunnel-api</artifactId>
        <version>2.3.2</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>com.google.auto.service</groupId>
        <artifactId>auto-service</artifactId>
        <version>1.0.1</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

添加一个 Java 类来实现 ZetaUDF，类似于以下的方式：

@AutoService(ZetaUDF.class)
public class ExampleUDF implements ZetaUDF {
    @Override
    public String functionName() {
        return "EXAMPLE";
    }

    @Override
    public SeaTunnelDataType<?> resultType(List<SeaTunnelDataType<?>> argsType) {
        return BasicType.STRING_TYPE;
    }

    @Override
    public Object evaluate(List<Object> args) {
        String arg = (String) args.get(0);
        if (arg == null) return null;
        return "UDF: " + arg;
    }
}

打包UDF项目并将jar文件复制到路径：${SEATUNNEL_HOME}/lib

示例

源端数据读取的表格如下：

id	name	age
1	Joy Ding	20
2	May Ding	21
3	Kin Dom	24
4	Joy Dom	22

我们使用SQL查询中的UDF来转换源数据，类似于以下方式：

transform {
  Sql {
    plugin_input = "fake"
    plugin_output = "fake1"
    query = "select id, example(name) as name, age from dual"
  }
}

那么结果表 fake1 中的数据将会更新为

id	name	age
1	UDF: Joy Ding	20
2	UDF: May Ding	21
3	UDF: Kin Dom	24
4	UDF: Joy Dom	22

更新日志

新版本

• 添加SQL转换连接器的UDF