Flink SQL Client注册Python UDF完整流程

港控/mmm° 2023-01-08 08:13 223阅读 0赞

### 概述 ###

本文是对\[1\]的复现

支持的版本是

<table style="width:478px;"> 
 <tbody> 
  <tr> 
   <td style="width:186px;"><strong>组件</strong></td> 
   <td style="width:130px;"><strong>版本</strong></td> 
   <td style="width:241px;"><strong>备注</strong></td> 
  </tr> 
  <tr> 
   <td style="width:186px;">Flink</td> 
   <td style="width:130px;">1.12.0</td> 
   <td style="width:241px;">apache版本</td> 
  </tr> 
  <tr> 
   <td style="width:186px;">apache-flink</td> 
   <td style="width:130px;">1.12.0</td> 
   <td style="width:241px;">pip安装</td> 
  </tr> 
  <tr> 
   <td style="width:186px;">python</td> 
   <td style="width:130px;">3.6.10</td> 
   <td style="width:241px;">anaconda虚拟环境</td> 
  </tr> 
 </tbody> 
</table>

**注意:**

①除了Flink集群要配置好，pip下面还要另外安装apache-flink(必须安装，不要遗漏).

②小版本号必须严格一致,例如下面这样:

Flink1.12.0与apache-flink1.11.2搭配使用是绝对不允许的.

### 完整代码与配置文件 ###

[https://github.com/appleyuchi/Flink\_SQL\_Client\_UDF][https_github.com_appleyuchi_Flink_SQL_Client_UDF]

### 完整操作步骤 ###

运行时采用的命令是:

$FLINK_HOME/bin/sql-client.sh embedded -e $FLINK_HOME/conf/env.yaml -pyfs $FLINK_HOME/conf/test.py

官网\[1\]中的写法如下:

# full example of python dependency management
    
    $ sql-client.sh embedded -e env.yaml -pyfs test.py,/home/my/func2.py -pyreq /home/my/requirements.txt#/home/my/pacakges_dir -pyarch py37.zip#venv -pyexec venv/py37/bin/python

使用下面命令查看已经注册func1:

show functions;

操作流程截图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70][]

实验结果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70 1][]

# 附录 #

可能遇到的报错:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70 2][]

Flink SQL> select func1(\`en\_behavior\`) as a from dim\_behavior;  
\[ERROR\] Could not execute SQL statement. Reason:  
java.lang.IllegalStateException: Process died with exit code 0

这是因为你的Flink集群版本和apache-flink版本不一致．

# **Reference** #

\[1\][FLIP-114: Support Python UDF in SQL Client][FLIP-114_ Support Python UDF in SQL Client]

[https_github.com_appleyuchi_Flink_SQL_Client_UDF]: https://github.com/appleyuchi/Flink_SQL_Client_UDF
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70]: /images/20221119/becde01104c342a3bb32b84e4a1563ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70 1]: /images/20221119/c518a288c3f74e7f91b1909f21089153.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FwcGxleXVjaGk_size_16_color_FFFFFF_t_70 2]: /images/20221119/a77162b2c3fa451a98eb42697786f0a6.png
[FLIP-114_ Support Python UDF in SQL Client]: https://cwiki.apache.org/confluence/display/FLINK/FLIP-114%3A+Support+Python+UDF+in+SQL+Client