流计算 Oceanus CREATE FUNCTION

这篇示例基于 Flink 1.20 编写，其他 Flink 版本的 UDF 写法基本相同，可直接参考；如个别 API 有差异，以对应版本的产品文档为准。

一、语法

目前流计算 Oceanus 支持 Java 和 Scala 两种语言编写的程序包。当用户上传了自定义程序包后，在界面上关联后即可用下面的 CREATE FUNCTION 语句来声明：

CREATE TEMPORARY SYSTEM FUNCTION 函数名 AS '函数类全名' [LANGUAGE JAVA|SCALA]

其中的函数名可以自行定义，但不要与现有的冲突。函数类全名为 Java 或 Scala 类的类全名（例如 'com.example.flink.MyCustomFunction'）。

二、示例

上传的程序包中包含类 com.example.flink.udf.MaskPhone，声明后即可在 SQL 中调用：

CREATE TEMPORARY SYSTEM FUNCTION mask_phone AS 'com.example.flink.udf.MaskPhone' LANGUAGE JAVA;SELECT user_id, mask_phone(phone) AS phone FROM user_log;

三、命名覆盖

如果存在系统内置的同名函数时，用户使用上述语法创建的 UDF 会覆盖系统内置的函数。因此除非有意改变系统函数的功能，请不要创建与系统内置函数同名的自定义函数。

四、示例

下面把自定义函数命名为 UPPER，声明之后当前作业里所有 UPPER(…) 都会走自定义实现，系统内置的 UPPER 不再生效：

CREATE TEMPORARY SYSTEM FUNCTION UPPER AS 'com.example.flink.udf.MyUpper' LANGUAGE JAVA;– 调用的是 com.example.flink.udf.MyUpper，而不是系统内置的 UPPERSELECT UPPER(name) FROM user_log;

五、函数类型

目前 Flink 支持下面多种函数定义。

六、标量函数（Scalar Function）

标量函数简称 UDF，作用是将一个值转换为另一个值（一对一），例如系统内置的 SUBSTRING、REPLACE 等字符串操作函数，都属于标量函数。

七、示例：手机号脱敏（一个输入值对应一个返回值）

实现时继承 org.apache.flink.table.functions.ScalarFunction，并实现公开的 eval 方法。eval 方法可以按入参类型重载多个。

package com.example.flink.udf;import org.apache.flink.table.functions.ScalarFunction;public class MaskPhone extends ScalarFunction { public String eval(String phone) { if (phone == null || phone.length() < 7) { return phone; } return phone.substring(0, 3) + "****" + phone.substring(7); }}

声明并调用：

CREATE TEMPORARY SYSTEM FUNCTION mask_phone AS 'com.example.flink.udf.MaskPhone' LANGUAGE JAVA;SELECT user_id, mask_phone(phone) AS phone FROM user_log;

输入输出示例：

八、表函数（Table Function）

表函数简称 UDTF，作用是将一个值转为表中的一行数据（一变多），这样可以在后续 JOIN 操作中作为右表。

九、示例：把逗号分隔的标签列拆成多行

继承 org.apache.flink.table.functions.TableFunction，泛型 T 是输出行的类型；在 eval 中调用 collect 输出零到多行。

package com.example.flink.udf;import org.apache.flink.table.functions.TableFunction;public class SplitTags extends TableFunction<String> { public void eval(String tags) { if (tags == null) { return; } for (String tag : tags.split(",")) { collect(tag); } }}

如果需要输出多列，把泛型换成 Row 并用注解声明输出类型：

package com.example.flink.udf;import org.apache.flink.table.annotation.DataTypeHint;import org.apache.flink.table.annotation.FunctionHint;import org.apache.flink.table.functions.TableFunction;import org.apache.flink.types.Row;@FunctionHint(output = @DataTypeHint("ROW<tag STRING, pos INT>"))public class SplitTagsWithPos extends TableFunction<Row> { public void eval(String tags) { if (tags == null) { return; } String[] array = tags.split(","); for (int i = 0; i < array.length; i++) { collect(Row.of(array[i], i)); } }}

声明并调用（用 LATERAL TABLE 做列转行）：

CREATE TEMPORARY SYSTEM FUNCTION split_tags AS 'com.example.flink.udf.SplitTags' LANGUAGE JAVA;SELECT user_id, tagFROM user_log, LATERAL TABLE(split_tags(tags)) AS T(tag);

输入输出示例：

十、聚合函数（Aggregate Function）

聚合函数简称 UDAGG，作用是将多行数据的一组值，聚合为一个最终值（多变一），例如系统内置的 MAX、MIN、AVG 等都属于聚合函数。

十一、示例：加权平均 weighted_avg

继承 org.apache.flink.table.functions.AggregateFunction，第一个泛型是返回值类型，第二个是累加器（accumulator）类型。必须实现

createAccumulator、accumulate、getValue；用到会话/滑动窗口聚合时还要实现 merge，用在 OVER 聚合或需要撤回的场景时还要实现

retract。

package com.example.flink.udf;import org.apache.flink.table.functions.AggregateFunction;public class WeightedAvg extends AggregateFunction<Double, WeightedAvg.WeightedAvgAccum> { public static class WeightedAvgAccum { public long weightedSum = 0L; public int weightSum = 0; } @Override public WeightedAvgAccum createAccumulator() { return new WeightedAvgAccum(); } public void accumulate(WeightedAvgAccum acc, Integer score, Integer weight) { if (score == null || weight == null) { return; } acc.weightedSum += score * (long) weight; acc.weightSum += weight; } public void retract(WeightedAvgAccum acc, Integer score, Integer weight) { if (score == null || weight == null) { return; } acc.weightedSum -= score * (long) weight; acc.weightSum -= weight; } public void merge(WeightedAvgAccum acc, Iterable<WeightedAvgAccum> iterable) { for (WeightedAvgAccum other : iterable) { acc.weightedSum += other.weightedSum; acc.weightSum += other.weightSum; } } @Override public Double getValue(WeightedAvgAccum acc) { if (acc.weightSum == 0) { return null; } return (double) acc.weightedSum / acc.weightSum; }}

声明并调用：

CREATE TEMPORARY SYSTEM FUNCTION weighted_avg AS 'com.example.flink.udf.WeightedAvg' LANGUAGE JAVA;SELECT student_id, weighted_avg(score, weight) AS avg_scoreFROM scoresGROUP BY student_id;

输入输出示例：

86.0 = (80 * 2 + 90 * 3) / (2 + 3)

十二、表聚合函数（Table Aggregate Function）

表聚合函数的作用是将多行数据的一组值，聚合为新的多行数据（多对多）。

十三、示例：求每个分组最大的两个值 top2

继承 org.apache.flink.table.functions.TableAggregateFunction，必须实现 createAccumulator、accumulate 和 emitValue（或 emitUpdateWithRetract）。

package com.example.flink.udf;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.table.functions.TableAggregateFunction;import org.apache.flink.util.Collector;public class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2.Top2Accum> { public static class Top2Accum { public Integer first; public Integer second; } @Override public Top2Accum createAccumulator() { return new Top2Accum(); } public void accumulate(Top2Accum acc, Integer value) { if (value == null) { return; } if (acc.first == null || value > acc.first) { acc.second = acc.first; acc.first = value; } else if (acc.second == null || value > acc.second) { acc.second = value; } } public void merge(Top2Accum acc, Iterable<Top2Accum> iterable) { for (Top2Accum other : iterable) { accumulate(acc, other.first); accumulate(acc, other.second); } } public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) { if (acc.first != null) { out.collect(Tuple2.of(acc.first, 1)); } if (acc.second != null) { out.collect(Tuple2.of(acc.second, 2)); } }}

表聚合函数在 Table API（Java/Scala 作业）中通过 flatAggregate 使用：

import static org.apache.flink.table.api.Expressions.$;import static org.apache.flink.table.api.Expressions.call;tableEnv.createTemporarySystemFunction("top2", Top2.class);tableEnv.from("orders") .groupBy($("category")) .flatAggregate(call("top2", $("price")).as("price", "rank")) .select($("category"), $("price"), $("rank")) .execute() .print();

输入输出示例：

十四、异步表函数（Async Table Function）

异步表函数可作为一种特殊的数据源，例如可以通过它来对接外部的数据库、数据存储。

十五、示例：异步查询 HTTP 维表服务

继承 org.apache.flink.table.functions.AsyncTableFunction，eval 的第一个参数是 CompletableFuture，异步结果通过 result.complete 返回，异常通过 result.completeExceptionally 抛出。open/close 用于初始化和释放外部客户端。

package com.example.flink.udf;import org.apache.flink.table.functions.AsyncTableFunction;import org.apache.flink.table.functions.FunctionContext;import org.apache.flink.types.Row;import org.apache.http.HttpResponse;import org.apache.http.client.methods.HttpGet;import org.apache.http.concurrent.FutureCallback;import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;import org.apache.http.impl.nio.client.HttpAsyncClients;import java.net.URLEncoder;import java.util.Collection;import java.util.Collections;import java.util.concurrent.CompletableFuture;public class DimQuery extends AsyncTableFunction<Row> { private transient CloseableHttpAsyncClient httpClient; private transient String endpoint; @Override public void open(FunctionContext context) throws Exception { endpoint = context.getJobParameter("dim.endpoint", "http://dim.service/query"); httpClient = HttpAsyncClients.createDefault(); httpClient.start(); } public void eval(CompletableFuture<Collection<Row>> result, String userId) { if (userId == null) { result.complete(Collections.emptyList()); return; } HttpGet request = new HttpGet(endpoint + "?user_id=" + URLEncoder.encode(userId, "UTF-8")); httpClient.execute( request, new FutureCallback<HttpResponse>() { @Override public void completed(HttpResponse response) { result.complete(parse(response)); } @Override public void failed(Exception ex) { result.completeExceptionally(ex); } @Override public void cancelled() { result.complete(Collections.emptyList()); } }); } private Collection<Row> parse(HttpResponse response) { // 解析响应，返回维表的零行、一行或多行 return Collections.singletonList(Row.of("user_name", "city")); } @Override public void close() throws Exception { if (httpClient != null) { httpClient.close(); } }}

异步表函数作为维表（Lookup）数据源的运行时实现，由连接器通过 AsyncTableFunctionProvider 提供：

import org.apache.flink.table.connector.source.AsyncTableFunctionProvider;import org.apache.flink.table.connector.source.LookupTableSource;public class HttpDimTableSource implements LookupTableSource { @Override public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { return AsyncTableFunctionProvider.of(new DimQuery()); }}

维表在 SQL 中以时态表 JOIN 的方式使用：

SELECT o.order_id, d.user_name, d.cityFROM orders AS oLEFT JOIN dim_user FOR SYSTEM_TIME AS OF o.proc_time AS dON o.user_id = d.user_id;

异步维表关联的并发、超时等参数在作业参数中设置：

table.exec.async-lookup.buffer-capacity = 100table.exec.async-lookup.timeout = 3min

十六、UDF 开发指南

由于 Flink 不同版本间 API 和文档迭代频繁，可参考 Flink 产品文档中的 UDF 开发指南。目前流计算 Oceanus 兼容开源版的 Flink 1.13、Flink1.14、Flink1.16、Flink1.18、Flink1.20 版本 API。