帮你快速理解、总结文档立即下载
文档中心>流计算 Oceanus>SQL 开发指南>DDL 数据定义语句>CREATE FUNCTION

CREATE FUNCTION

最近更新时间：2026-10-27 18:44:50
我的收藏

本页目录：

对于 SQL 作业，用户可以上传 自定义程序包，然后在作业分析开发页的参数设置中，引用该程序包。目前支持从本地上传，也可以引用账户下现有 COS 存储中的资源（仅限相同地域）。这里的程序包既可以用来扩展 Connector 的功能，也可以创建自定义函数（UDF）。

本文示例基于 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，而不是系统内置的 UPPER
SELECT 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;
输入输出示例：
phone
mask_phone(phone)
13812345678
138****5678
13900001111
139****1111
null
null

表函数（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, tag
FROM user_log, LATERAL TABLE(split_tags(tags)) AS T(tag);
输入输出示例：
输入 user_log
输出
(1001, 'vip,new,sz')
(1001, 'vip')

(1001, 'new')

(1001, 'sz')

聚合函数（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_score
FROM scores
GROUP BY student_id;
输入输出示例：
输入 scores
输出
(1, 80, 2) (1, 90, 3)
(1, 86.0)
(2, 70, 1) (2, 60, 1)
(2, 65.0)
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();
输入输出示例：
输入 orders
输出
('电子', 199) ('电子', 299) ('电子', 99) ('图书', 45)
('电子', 299, 1)

('电子', 199, 2)

('图书', 45, 1)

异步表函数（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.city
FROM orders AS o
LEFT JOIN dim_user FOR SYSTEM_TIME AS OF o.proc_time AS d
ON o.user_id = d.user_id;
异步维表关联的并发、超时等参数在作业参数中设置：
table.exec.async-lookup.buffer-capacity = 100
table.exec.async-lookup.timeout = 3min


UDF 开发指南

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