概述

听大佬说[1]里面有flink sql client注册udf的方法

去看了一眼，全是文字，闹心，索性琢磨了一下，记录下来．

 

UDF的完整maven工程

https://github.com/appleyuchi/Flink_SQL_Client_UDF

 

完整操作步骤

①mvn clean package

或者直接点击intellij右侧的package按钮来进行编译

②编译得到的jar放入到集群的下面，最终路径如下(不一定非要下面这个路径，也可以是其他你自己指定的路径):
$FLINK_HOME/lib/table_api-1.0-SNAPSHOT.jar

③下面这步有说法是zeppelin才需要,sql-client不需要

$FLINK_HOME/conf/flink-conf.yaml中末尾加入这么一句

flink.execution.jars: /home/appleyuchi/bigdata/flink-1.12/lib/table_api-1.0-SNAPSHOT.jar

上面的jar只要放到你启动flink sql client 的那个节点中的$FLINK_HOME/lib下面即可，不需要同步到其他节点

不要问如果我有多个udf的jar怎么办，

凉拌，

自己去把所有udf整合到一个jar中来．

 

④启动Flink SQL Client

输入

create temporary function hash_compute as 'Hashstring';

如果你是先启动Flink SQL Client,后把jar拷贝到②中，那么必须重启Flink sql client，否则会识别不到该jar。

⑤定义source

CREATE TABLE orders (order_uid BIGINT,product_name String,price DECIMAL(32, 2),order_time TIMESTAMP(3) ) WITH ('connector' = 'datagen' );SELECT * FROM orders;

不用担心输入的数据是多少，上面的datagen随机给你生成．

⑥调用UDF

select hash_compute(product_name) from orders;

 

附录

操作过程截图

 

实验效果

 

可能遇到的报错

[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: Hashstring
这是因为，这个jar是你启动flink集群后才拷贝过去的．

Flink SQL Client加载包含udf的table_api-1.0-SNAPSHOT.jar只有两种办法:

①table_api-1.0-SNAPSHOT.jar拷贝到$FLINK_HOME/lib下面后重启Flink集群

②$FLINK_HOME/bin/sql-client.sh embedded -j $FLINK_HOME/lib/table_api-1.0-SNAPSHOT.jar

 

 

Reference:

[1]CREATE Statements

[2]UDF

 

总结

以上是生活随笔为你收集整理的Flink SQL Client注册JAVA UDF完整流程的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。