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）

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

-- -------------------- ---- -------
------ -----------------------------------------------
------ ---------------------------

------ ----- --------------- ------- ------------------ -
    ------ ---- ----------- ------ -
        --- ------- - - ----------------- -
            -------------------
        -
    -
-

聚合函数（Aggregate Function）

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

-- -------------------- ---- -------
------ ---------------------------------------------------

------ ----- ------------------- ------- ----------------------- -------------- -
    ------ ------------- ------------------- -
        ------ --- ----------------
    -

    ------ ---- ---------------------- ------------ -
        ------ ----------------
    -

    ------ ---- ------------------------ ------------ ---- ------ -
        --------------- -- ------
    -

    ------ ------ ----- ------------- -
        ------ ---- --- - --
    -
-

2. 注册 UDF

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

-- -------------------- ---- -------
---------------------- -------- - -----------------------------------

-- ------
----------------------------------------- --- --------------------

-- -----
---------------------------------------- --- -------------------

-- ------
-------------------------------------- --- -----------------------

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

-- -------------------- ---- -------
----- ------ - ----------------------
    ---------------------------- ------------

----- ------- - ----------------------
    -------------------------------- -----------------------------
    ------------------ -----------

----- ------- - ----------------------
    -------------------
    ------------------ ----------------- -----------

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