如何在PyFlink中导入私有包并在自定义UDF中调用其功能

方案1：在requirements.txt中配置私有仓库源 #

你可以直接在requirements.txt中声明私有源地址，继续使用set_python_requirements接口，无需修改其他配置：

1. 在requirements.txt首行添加私有仓库配置，需要认证的话可以把认证信息拼接在地址中：

--extra-index-url https://<用户名>:<密码>@你的私有pip仓库地址/simple
你的私有包名==版本号
其他公共依赖包名==版本号

2. 正常调用原接口配置依赖即可：

table_env.set_python_requirements(requirements_file_path="/path/to/requirements.txt")

Flink提交作业时会自动从配置的私有源拉取对应包分发到各个执行节点。

方案2：离线上传预下载的私有包 #

如果作业运行环境无法访问私有仓库，可以提前把私有包下载到本地后随作业一起上传：

• 单私有包场景：直接上传wheel或源码包
  下载私有包的.whl安装包或.tar.gz源码包到本地，通过add_python_file接口上传，Flink会自动将包加入各节点的Python搜索路径：

  table_env.add_python_file("/path/to/私有包文件.whl")
  

  作业中直接import包即可正常使用。
• 多依赖/复杂依赖树场景：打包整个Python虚拟环境
  1. 在本地准备和集群Python版本完全一致的虚拟环境，手动安装好所有依赖（包括私有包）
  2. 将整个虚拟环境打包为zip压缩包
  3. 通过以下配置指定用该虚拟环境运行作业：

  # 第二个参数的#后面的别名可以自定义，和下面的解释器路径保持一致即可
  table_env.set_python_archives("venv.zip#venv")
  table_env.get_config().set_python_executable("venv/bin/python")
  

方案3：集群预安装依赖 #

如果是固定维护的Flink集群，可以直接在所有JobManager、TaskManager节点的默认Python环境中提前安装好该私有包，作业代码无需做任何额外依赖配置，直接import即可正常调用。

注意事项 #

• 本地开发环境的Python大版本必须和集群运行环境保持一致，避免出现依赖兼容性问题
• 若私有包依赖系统级动态库，需要提前在所有集群节点安装对应的系统依赖
• 打包虚拟环境时建议删除__pycache__、pip缓存等冗余文件，缩小压缩包体积提升提交效率

内容的提问来源于stack exchange，提问作者ElCapitaine