Flink 中 Table API & SQL 如何使用 UDF (用户自定义函数)？

推荐答案

在 Flink 中，Table API 和 SQL 支持使用 UDF（用户自定义函数）来扩展内置函数的功能。以下是使用 UDF 的步骤：

1. 定义 UDF 类：

   • 对于标量函数（Scalar Function），继承 ScalarFunction 类并实现 eval 方法。
   • 对于表函数（Table Function），继承 TableFunction 类并实现 eval 方法。
   • 对于聚合函数（Aggregate Function），继承 AggregateFunction 类并实现相关方法。

2. 注册 UDF：

   • 使用 TableEnvironment 的 registerFunction 方法将 UDF 注册到 Flink 中。

3. 在 SQL 或 Table API 中使用 UDF：

   • 在 SQL 查询中直接调用注册的 UDF。
   • 在 Table API 中使用 call 方法来调用 UDF。

本题详细解读

1. 定义 UDF 类

标量函数（Scalar Function）

标量函数是最常见的 UDF 类型，它将零个或多个标量值映射到一个标量值。

import org.apache.flink.table.functions.ScalarFunction;

public class MyScalarFunction extends ScalarFunction {
    public String eval(String input) {
        return input.toUpperCase();
    }
}

表函数（Table Function）

表函数可以将一行数据转换为多行数据。

// www.javascriptcn.com code example
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

public class MyTableFunction extends TableFunction<Row> {
    public void eval(String input) {
        for (String s : input.split(",")) {
            collect(Row.of(s));
        }
    }
}

聚合函数（Aggregate Function）

聚合函数用于对一组数据进行聚合操作。

// www.javascriptcn.com code example
import org.apache.flink.table.functions.AggregateFunction;

public class MyAggregateFunction extends AggregateFunction<Long, MyAccumulator> {
    public MyAccumulator createAccumulator() {
        return new MyAccumulator();
    }

    public Long getValue(MyAccumulator accumulator) {
        return accumulator.sum;
    }

    public void accumulate(MyAccumulator accumulator, long value) {
        accumulator.sum += value;
    }

    public static class MyAccumulator {
        public long sum = 0;
    }
}

2. 注册 UDF

在 Flink 中，UDF 需要先注册到 TableEnvironment 中才能使用。

// www.javascriptcn.com code example
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 注册标量函数
tableEnv.registerFunction("myScalarFunc", new MyScalarFunction());

// 注册表函数
tableEnv.registerFunction("myTableFunc", new MyTableFunction());

// 注册聚合函数
tableEnv.registerFunction("myAggFunc", new MyAggregateFunction());

3. 在 SQL 或 Table API 中使用 UDF

在 SQL 中使用 UDF

SELECT myScalarFunc(name) FROM users;
SELECT * FROM LATERAL TABLE(myTableFunc(description)) AS T(word);
SELECT myAggFunc(age) FROM users GROUP BY name;

在 Table API 中使用 UDF

// www.javascriptcn.com code example
Table result = tableEnv.from("users")
    .select(call("myScalarFunc", $("name")));

Table result2 = tableEnv.from("users")
    .joinLateral(call("myTableFunc", $("description")).as("word"))
    .select($("name"), $("word"));

Table result3 = tableEnv.from("users")
    .groupBy($("name"))
    .select($("name"), call("myAggFunc", $("age")));

通过以上步骤，你可以在 Flink 的 Table API 和 SQL 中成功使用 UDF 来扩展数据处理能力。