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;
本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。
