数据处理的逻辑越来越复杂，SQL 也越写越长，有时会觉得 SQL 力不从心。这时就需要扩展 SQL 的表达能力，用户自定义函数（UDF）正是这样一种机制：在引擎内置函数之外，接入自己写的逻辑。

这篇文章从三个具体问题出发，分别用一种 UDF 解决：

问题	用到的 UDF	一句话
ETL 中同一个正则写了四遍	表值函数（UDTF）	一次解析，输出多列
离线计算「当天截至每分钟的累计 UV」	聚合函数（UDAF）	把 HyperLogLog 引入 SQL
实时流里调用机器学习模型	Python UDF	接入 Python 生态

先认识三类 UDF

类型	输入 → 输出	内置函数的例子
标量函数（UDF）	一行的若干值 → 一个值	`DATE_FORMAT`
表值函数（UDTF）	一行的若干值 → 若干行，每行可有多列	`EXPLODE`
聚合函数（UDAF）	多行 → 一个值	`SUM`

在 Flink 里自定义函数的步骤大致相同（详见官方文档）：

继承对应的基类（ScalarFunction、TableFunction、AggregateFunction）；
实现计算方法（eval、accumulate 等），特别注意输入输出的类型推导；
打包后在 SQL 里注册，就能像内置函数一样调用。

文中只给出关键代码，完整实现见 QFlink 的 qflink-sql-udf。下面的例子接着上一篇的数据，先准备好日志表：

shell

tutorials/run.sh sql_is_all_your_need/prepare.sql

表值函数：一次解析，输出多列

上一篇留了一个问题：ETL 里同一个正则写了四遍。

sql

INSERT INTO `dwd_log`
SELECT
    REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 1) AS `time`,
    REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 2) AS `key_word`,
    JSON_VALUE(REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 3), '$.key1' RETURNING INT) AS `key1`,
    JSON_VALUE(REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 3), '$.key2') AS `key2`
FROM `ods_log`;

冗长的根本原因是：标量函数一次只能返回一个值。正则有三个分组，就得调用三次；JSON 里要取两个字段，又得各解析一次。

表值函数可以返回多列，正好解决这个问题。配合 LATERAL TABLE，左边表的每一行都会和函数为它返回的行拼接起来，一次调用就能拆出所有字段：

表值函数一次拆出所有字段：正则只执行一次，JSON 只解析一次

sql

CREATE TEMPORARY FUNCTION IF NOT EXISTS q_regex_extract
    AS 'cn.syntomic.qflink.sql.udf.table.QRegexExtract' LANGUAGE JAVA;

SELECT `time`, `key_word`, `json`
FROM `ods_log`,
    LATERAL TABLE(q_regex_extract(`log`,
        '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)',
        'time', 'key_word', 'json'));

JSON 里的字段用 Flink 自带的 JSON_VALUE 取出。最初这里用的是 Hive 模块的 json_tuple，同样能一次取出多个字段；但 Hive 连接器最新只发布到 Flink 1.20，QFlink 升级到 Flink 2.2 之后就加载不了了。完整的清洗就是这样（flink_sql_udf/etl_udtf.sql）：

sql

INSERT INTO `dwd_log`
SELECT
    `time`,
    `key_word`,
    JSON_VALUE(`json`, '$.key1' RETURNING INT) AS `key1`,
    JSON_VALUE(`json`, '$.key2') AS `key2`
FROM
    `ods_log`, LATERAL TABLE(q_regex_extract(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 'time', 'key_word', 'json'));

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql_udf/etl_udtf.sql
tutorials/run.sh --batch tutorials/show.sql --table dwd_log --columns '`time`, key_word, key1, key2'

text

+----+---------------------+----------+------+------+
| op |                time | key_word | key1 | key2 |
+----+---------------------+----------+------+------+
| +I | 2023-04-09 15:40:05 |   stdout |    5 | val1 |
| +I | 2023-04-09 15:41:05 |   stdout |    4 | val2 |
| +I | 2023-04-09 15:42:05 |   stdout |    1 | val2 |
| +I | 2023-04-09 15:43:05 |   stdout |    5 | val3 |
| +I | 2023-04-09 15:44:05 |   stdout |    6 | val4 |
+----+---------------------+----------+------+------+

结果和上一篇完全一样，区别在计算量：正则从每行四次变成了一次。

这样清洗逻辑只写一遍，也顺带避开了 Flink 中同一个 UDF 被重复调用的问题。

让函数自己说出列名

细心的话会发现，q_regex_extract 后面不用写 AS t(time, key_word, json) 这样的别名，而 Hive 的 json_tuple 这类函数必须写。区别在于 q_regex_extract 自定义了类型推导：它把调用时传入的字符串常量（'time'、'key_word'、'json'）直接当作输出列的名字。

java

@Override
public TypeInference getTypeInference(DataTypeFactory typeFactory) {
    return TypeInference.newBuilder()
            .outputTypeStrategy(callContext -> {
                int argsLen = callContext.getArgumentDataTypes().size();
                List<Field> fields = new ArrayList<>(argsLen - 2);
                // 前两个参数是原文和正则，之后每个参数都是一个输出列名
                for (int i = 2; i < argsLen; i++) {
                    String fieldName = callContext
                            .getArgumentValue(i, String.class)
                            .orElse("f" + (i - 2));
                    fields.add(DataTypes.FIELD(fieldName, DataTypes.STRING()));
                }
                return Optional.of(DataTypes.ROW(fields.toArray(new Field[0])));
            })
            .build();
}

类型推导发生在 SQL 编译期：优化器根据这里返回的类型，就知道这个函数会产出哪几列、叫什么名字。

聚合函数：把 HyperLogLog 引入 SQL

上一篇按一分钟的滚动窗口统计了去重数。现在把问题变难一点：每分钟输出一次「当天 0 点到现在」的累计去重数。

累积窗口的起点固定、终点每分钟往后推，每个窗口都包含前面所有的数据（图里起点画在 15:40，实际是当天 0 点）

Flink 1.13 之后，流模式下可以直接用累积窗口 CUMULATE：

sql

SELECT
    DATE_FORMAT(`window_start`, 'yyyy-MM-dd HH:mm:ss') AS `window_start`,
    DATE_FORMAT(`window_end`, 'yyyy-MM-dd HH:mm:ss') AS `window_end`,
    `key_word` AS `dim`,
    COUNT(DISTINCT `key2`) AS `metric`
FROM TABLE(
    CUMULATE(TABLE `dwd_log`, DESCRIPTOR(`rowtime`), INTERVAL '1' MINUTES, INTERVAL '1' DAYS))
GROUP BY `key_word`, `window_start`, `window_end`;

但当时（Flink 1.15）的批模式还不支持累积窗口去重。离线要怎么算？

最直接的思路是：先列出当天所有的分钟窗口，再把每条数据关联到所有结束时间比它晚的窗口上，按窗口去重。问题是一条数据会被复制到几百上千个窗口，数据量一大就会膨胀、倾斜。可以先做一些优化：

累计去重只关心每个值当天第一次出现的时间，先按值取最早时间，数据量能少很多；
调整资源和参数，打开自适应执行。

更好的办法是换一种数据结构。HyperLogLog（HLL）是一种基数估计算法，用约 1.2 KB 内存就能以 2% 左右的误差估计上亿的基数，而且两个 HLL 可以合并。于是思路变成：

每分钟的数据聚合成一个 HLL；
按时间顺序，把当天第一分钟到当前分钟的 HLL 依次合并，再估计基数。

下图中的数字只是示意：

每分钟一个 HLL，按时间顺序累积合并，得到从 0 点到现在的去重数（数字为示意）

第一步用一个聚合函数 hll_agg 实现，第二步用窗口函数 OVER (ORDER BY ...) 做「从第一行到当前行」的累积合并：

sql

CREATE TEMPORARY FUNCTION IF NOT EXISTS hll_agg
    AS 'cn.syntomic.qflink.sql.udf.aggregate.HLLAggregate' LANGUAGE JAVA;
CREATE TEMPORARY FUNCTION IF NOT EXISTS hll_cardinality
    AS 'cn.syntomic.qflink.sql.udf.scalar.HLLCardinality' LANGUAGE JAVA;

SELECT
    DATE_FORMAT(`min_time`, 'yyyy-MM-dd 00:00:00') AS `window_start`,
    FROM_UNIXTIME(UNIX_TIMESTAMP(`min_time`) + 60) AS `window_end`,
    `key_word` AS `dim`,
    -- 按分钟排序，合并第一行到当前行的 HLL，再估计基数
    hll_cardinality(hll_agg(min_hll) OVER (PARTITION BY `key_word` ORDER BY `min_time`)) AS `metric`
FROM (
    -- 每分钟的数据聚合成一个 HLL
    SELECT
        DATE_FORMAT(`time`, 'yyyy-MM-dd HH:mm:00') AS `min_time`,
        `key_word`,
        hll_agg(`key2`) AS min_hll
    FROM `dwd_log`
    GROUP BY DATE_FORMAT(`time`, 'yyyy-MM-dd HH:mm:00'), `key_word`
) a;

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql_udf/analysis_hll.sql

text

+----+---------------------+---------------------+--------+--------+
| op |        window_start |          window_end |    dim | metric |
+----+---------------------+---------------------+--------+--------+
| +I | 2023-04-09 00:00:00 | 2023-04-09 15:41:00 | stdout |      1 |
| +I | 2023-04-09 00:00:00 | 2023-04-09 15:42:00 | stdout |      2 |
| +I | 2023-04-09 00:00:00 | 2023-04-09 15:43:00 | stdout |      2 |
| +I | 2023-04-09 00:00:00 | 2023-04-09 15:44:00 | stdout |      3 |
| +I | 2023-04-09 00:00:00 | 2023-04-09 15:45:00 | stdout |      4 |
+----+---------------------+---------------------+--------+--------+

和上图对得上：第三分钟又来了一次 val2，累计值停在 2。能这样逐分钟合并，是因为 HyperLogLog 能合并；但它不能撤销，所以近 N 天的 UV 不能用「减掉 N 天前那天」的递推，只能把 N 份合并，原因见少算一点：大数据计算优化的统一视角。

hll_agg 既能聚合原始值，也能合并 HLL，所以两处都用它。HLL 不是 Flink 内置的数据类型，聚合函数的累加器里要把它声明为 RAW 类型，由 Kryo 序列化：

java

public class HLLBuffer {

    @DataTypeHint(allowRawGlobally = HintFlag.TRUE)
    public HLL hll = new HLL(11, 5);
}

想更直观地理解 HLL，可以读神奇的 HyperLogLog。借助 UDAF，SQL 里就多了一种全新的数据结构。

Python UDF：把模型接进实时流

有时需要借助 Python 生态，比如动态执行脚本、调用机器学习模型做预测。Flink 支持用 Python 写 UDF，下面用经典的鸢尾花分类做例子。

原理：两种执行模式

Java 算子和 Python 函数怎么协作？Flink 提供了两种模式：

进程模式：Python 函数跑在一个独立的 Python 进程里，和 Java 算子通过 gRPC 交换数据。隔离性好，但每条数据都要跨进程序列化。

进程模式：Python 在另一个进程里，每条数据都要序列化、跨进程来回一趟

线程模式：Python 解释器嵌入 Java 进程，两边通过外部函数接口（FFI）直接调用，省掉了进程间的序列化开销，性能更好。其中的跨语言调用原理，见 Java 调用 Python。

线程模式：Python 解释器嵌在同一个 JVM 里，省掉了跨进程的序列化，代价是共用一把 GIL

实现

先写一个 Python 类：open 在任务启动时加载一次模型，eval 对每条数据做预测。

python

class DemoUDF(ScalarFunction):

    def open(self, function_context: FunctionContext):
        self.model = joblib.load(path)

    def eval(self, value):
        return self.model.predict(np.array(value).reshape(1, -1)).tolist()[0]


demo_udf = udf(DemoUDF(), result_type=DataTypes.INT())

然后在 SQL 里注册并调用：

sql

SET 'python.files' = './qflink-python/src/main/python/q-pyflink/udf';
-- 断开算子链，让 Python 算子获得足够的资源
SET 'pipeline.operator-chaining' = 'false';
-- 线程模式
SET 'python.execution-mode' = 'thread';

CREATE TEMPORARY FUNCTION IF NOT EXISTS demo_udf AS 'scalar.demo_udf.demo_udf' LANGUAGE PYTHON;

CREATE TABLE IF NOT EXISTS `source` (
    `sepal_length` FLOAT,
    `sepal_width` FLOAT,
    `petal_length` FLOAT,
    `petal_width` FLOAT
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '1'
);

SELECT
    demo_udf(ARRAY[`sepal_length`, `sepal_width`, `petal_length`, `petal_width`]) AS `classifier`
FROM `source`;

运行前先准备 QFlink 的 Python 环境（只需一次），再加 --python 运行。这是流作业，datagen 每秒造一条鸢尾花数据，看到几行输出后按 Ctrl-C：

shell

(cd qflink-python/src/main/python/q-pyflink && uv sync)
tutorials/run.sh --python sql_is_all_your_need/flink_sql_udf/python_udf.sql

text

+I[0]
+I[2]
+I[0]
+I[2]
+I[1]

每行是模型给出的类别，数据是随机生成的，你看到的序列会不一样。--python 做了两件事：把 uv 环境里的 python 放进 PATH，线程模式靠它找到 libpython；把环境的 site-packages 放进 PYTHONPATH，嵌进 JVM 的解释器从这里导入 pemja 和模型依赖。少了任何一个，作业都会在启动后失败。如果 Python 算子是 CPU 密集型的，还要相应调高 TaskManager 的 CPU 数。

总结

大数据技术日新月异，但 SQL never dies。这篇文章展示了三种用 UDF 拓宽 SQL 边界的方式：UDTF 让一次调用产出多列，UDAF 把新的数据结构带进 SQL，Python UDF 把整个 Python 生态接进来。

技术没有银弹，SQL 也不能解决所有问题。SQL 和底层 API 都只是解决问题的工具，要根据问题选择合适的那个，甚至可以把 SQL 的通用和 DataStream API 的灵活结合起来。

本文的全部 SQL 在 QFlink，运行说明见 QFlink 教程第 02 章。

参考资料

全文评论

正在加载评论…