UDF的完整maven工程與SQL

https://github.com/appleyuchi/Flink_SQL_Client_UDF

?

完整操作步驟

①mvn scala:compile package

?

②編譯得到的jar放入到集群的下面，最終路徑如下(不一定非要下面這個路徑，也可以是其他你自己指定的路徑):

$FLINK_HOME/lib/flink-udf-1.0-SNAPSHOT.jar

?

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

? flink.execution.jars: $FLINK_HOME/lib/flink-udf-1.0-SNAPSHOT.jar ?

上面的jar只要放到你啟動flink sql client 的那個節點中的$FLINK_HOME/lib下面即可，不需要同步到其他節點

不要問如果我有多個udf的jar怎么辦，

涼拌，

自己去把所有udf整合到一個jar中來．

?

④啟動Flink SQL Client

輸入

create temporary function scalaupper as 'org.apache.zeppelin.flink.udf.ScalaUpper';

如果你是先啟動Flink SQL Client,后把jar拷貝到②中，那么必須重啟Flink sql client，否則會識別不到該jar。

?

⑤Flink SQL Client中定義source

不用擔心輸入的數據是多少，SQL中的datagen隨機給你生成．

具體的定義source的SQL見上面的git

⑥調用UDF

select scalaupper(product_name) from orders;

操作過程截圖

?

實驗結果截圖

?

可能遇到的報錯

[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: Hashstring
這是因為，這個jar是你啟動flink集群后才拷貝過去的．

Flink SQL Client加載包含udf的flink-udf-1.0-SNAPSHOT.jar只有兩種辦法:

①flink-udf-1.0-SNAPSHOT.jar拷貝到$FLINK_HOME/lib下面后重啟Flink集群

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

?

?

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。