概述

聽大佬說[1]里面有flink sql client注冊udf的方法

去看了一眼，全是文字，鬧心，索性琢磨了一下，記錄下來．

?

UDF的完整maven工程

https://github.com/appleyuchi/Flink_SQL_Client_UDF

?

完整操作步驟

①mvn clean package

或者直接點擊intellij右側(cè)的package按鈕來進(jìn)行編譯

②編譯得到的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 的那個節(jié)點中的$FLINK_HOME/lib下面即可，不需要同步到其他節(jié)點

不要問如果我有多個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;

不用擔(dān)心輸入的數(shù)據(jù)是多少，上面的datagen隨機(jī)給你生成．

⑥調(diào)用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

?

總結(jié)

以上是生活随笔為你收集整理的Flink SQL Client注册JAVA UDF完整流程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。