Flink 使用介绍相关文档目录

Flink 使用介绍相关文档目录

前言

接上篇Flink 使用之 SQL UDF。本篇主要分享如何实现UDF统一管理和使用时加载的方式。

构建UDF jar包

将UDF和作业jar包分离可以实现UDF的统一管理。这些UDF还可以在不同的作业之间复用，提高了灵活性和易用性。因此建议大家都这么做。

编写UDF jar包必须的Flink依赖仅有这一个：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

引入这个依赖之后可以编写UDF。下面以字符串转换大小写的UDF为例：

public class ChangeCaseTool extends ScalarFunction {
    public String eval(String s) {
        return s.toUpperCase(Locale.ROOT);
    }
    public String eval(String s, Boolean changeToUppercase) {
        if (changeToUppercase) {
            return s.toUpperCase(Locale.ROOT);
        } else {
            return s.toLowerCase(Locale.ROOT);
        }
    }
}

编写完毕之后执行maven package将UDF打包。编译输出的jar文件复制到服务器备用。

同理，经常使用到的逻辑都可以使用这种方式，制作成一个个的UDF jar包，合在一起形成了UDF库。在作业中按需加载和调用。

加载UDF的方式

目前试验可行的加载UDF方法如下：

方法1：启动Flink之前或提交Yarn任务之前，将UDF jar包放入Flink安装目录的lib中。这种方式无论提交什么作业都会加载lib中的所有UDF，灵活性欠佳。

方法2：对于SQL Client，启动的时候使用-j参数指定UDF jar包。例如：

./sql-client.sh -s embedded -j /path/to/udf.jar

方法3：对于提交jar作业的方式，使用-C参数指定UDF jar文件。同时使用-yt参数将jar传输到Yarn集群。-C和-yt指定的路径要保持一致。

./flink run -m yarn-cluster -yt /path/to/udf.jar -C file:///path/to/udf.jar /path/to/flink-job.jar 

方法4：SQL语句使用ADD JAR动态载入UDF jar。仅限于Flink 1.14.x及其之后的版本才可以使用。

添加保存在HDFS中的UDF jar：

add jar 'hdfs:///path/to/udf.jar

添加存放在本地的UDF jar：

add jar '/path/to/udf.jar';

添加成功之后，使用show jars检查jar文件是否已加载：

Flink SQL> show jars;
+---------------------------------------------------+
|                                              jars |
+---------------------------------------------------+
| hdfs:///path/to/udf.jar                           |
+---------------------------------------------------+
1 row in set

SQL JAR语句官网链接：https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/table/sql/jar/

SQL中使用UDF

下面以一个使用datagen随机生成string类型的name字段的表为例，介绍下UDF如何加载调用。

// 样例表，随机生成string
create table demo (name string) with ('connector' = 'datagen');

// 创建change_case UDF，逻辑位于com.paultech.ChangeCaseTool，对应UDF jar中类的全名
create temporary function change_case as 'com.paultech.ChangeCaseTool';

// 调用UDF
select change_case(name, true) from demo limit 5;

本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。