PyFlink 依赖配置指南：JAR、Python 包与虚拟环境集成

B）add_classpaths(...)：加到 client + cluster classpath（同样要求可达）

env.add_classpaths("file:///opt/flink/jars/connector1.jar","file:///opt/flink/jars/connector2.jar")

2.3 提交参数 --jarfile 的限制

• 只支持一个 jar，所以多个依赖通常要求你自己打 fat jar / uber jar

3. Python 依赖：三种层级（文件/目录、requirements、归档环境）

3.1 python.files / add_python_file：带'代码/包'到 PYTHONPATH

适合：

• 你的 UDF 写在 my_udf.py
• 或者你有一坨自研包目录 my_pkg/
• 或者你打好了 *.whl / *.egg（zip 本质）

Table API：

table_env.add_python_file("/path/to/my_udf.py")
table_env.add_python_file("/path/to/my_pkg/") # 目录也可以

DataStream API：

env.add_python_file("/path/to/my_udf.py")
env.add_python_file("/path/to/my_pkg/")

等价的还有：

• 配置 python.files
• 提交参数 -pyfs / --pyFiles

关键点：这些会被加到 Python UDF worker 的 PYTHONPATH。

3.2 requirements.txt / set_python_requirements：让集群 pip 安装第三方依赖

适合：

• numpy/pandas/requests/sklearn 这种 pip 依赖
• 你希望 Flink 在 worker 上自动安装

Table API：

table_env.set_python_requirements(
    requirements_file_path="/path/to/requirements.txt",
    requirements_cache_dir="cached_dir" # 可选
)

DataStream API：

env.set_python_requirements(
    requirements_file_path="/path/to/requirements.txt",
    requirements_cache_dir="cached_dir"
)

离线安装（集群没网）怎么做？

文档给了关键命令：

pip download -d cached_dir -r requirements.txt --no-binary :all:

然后把这个 cached_dir 作为 requirements_cache_dir 传进去，Flink 会上传它用于离线安装。

硬要求（很容易忽视）：

• pip >= 20.3
• setuptools >= 37.0.0
• cached_dir 里的包必须匹配集群平台与 Python 版本（比如 manylinux、glibc、cp310/cp311）

等价的还有：

• 配置 python.requirements
• 提交参数 -pyreq / --pyRequirements

3.3 python.archives / add_python_archive：带'环境/数据/模型文件'并自动解压

适合：

• 你要带模型、词典、数据文件
• 你要带一个完整虚拟环境（venv/conda 打包）

Table API：

table_env.add_python_archive("/path/to/py_env.zip","myenv")

DataStream API：

env.add_python_archive("/path/to/py_env.zip","myenv")

在 UDF 里访问（相对路径）：

def my_udf():
    with open("myenv/py_env/data/data.txt") as f:
        ...

如果没写 target_dir：

table_env.add_python_archive("/path/to/py_env.zip") # UDF 内访问：open("py_env.zip/py_env/data/data.txt")

支持格式：

• zip 系（zip/jar/whl/egg…）
• tar 系（tar/tar.gz/tgz…）

注意：如果 archive 里是虚拟环境，一定要和集群平台一致。

4. Python 解释器：worker 端与 client 端是两回事

这是很多人线上翻车的根本原因： client 侧需要 Python 来解析/编译 UDF；cluster 侧 worker 需要 Python 来执行 UDF。

4.1 worker 端 Python：python.executable / set_python_executable

Table API：

table_env.get_config().set_python_executable("/path/to/python")

DataStream API：

env.set_python_executable("/path/to/python")

解释器放在 archive 里（推荐'自带环境'打法）

env.add_python_archive("/path/to/py_env.zip","venv")
env.set_python_executable("venv/py_env/bin/python")

注意：如果指向 archive 内路径，用 相对路径，别写绝对路径。

等价方式：

• 配置 python.executable
• 提交参数 -pyexec / --pyExecutable

4.2 client 端 Python：python.client.executable / --pyClientExecutable

client 端用于'编译阶段解析 Python UDF'。你可以：

• 直接激活你本地的 venv：source my_env/bin/activate
• 或通过配置/参数指定：python.client.executable / -pyclientexec / PYFLINK_CLIENT_EXECUTABLE

5. 在 Java/SQL 里用 Python UDF：依赖还是走 Python 那套配置

你可以在 Java Table API 或纯 SQL 里注册 Python UDF，例如：

tEnv.executeSql("create temporary system function add_one as 'add_one.add_one' language python");

但 Python 依赖依然要通过这些配置/参数提供：

• python.files
• python.archives
• python.requirements
• python.executable
• python.client.executable

6. 一套工程化推荐组合（拿去就能用）

场景 A：集群能上网（最省事）

• JAR：pipeline.jars 或 env.add_jars
• Python：set_python_requirements(requirements.txt)
• 自研代码：add_python_file(my_udf.py / my_pkg/)
• 模型文件：add_python_archive(model.zip, "model")

场景 B：集群没网（企业内网最常见）

• requirements 离线缓存：pip download -d cached_dir -r requirements.txt
• 代码依赖：add_python_file(...)
• 模型/数据：add_python_archive(...)
• 如果集群 Python 环境不可控：直接把 venv 打包进 archive，再 set_python_executable("venv/.../python")

场景 C：你要'零环境依赖'的可移植作业（最稳）

• 把 venv 打包成 zip（或 conda pack）
• add_python_archive(venv.zip, "venv")
• set_python_executable("venv/.../python")
• 第三方包不再走 requirements（除非你愿意让它再装一遍）

7. 最常见的坑清单（提前避雷）

• 混用 Table + DataStream，你只在 Table 侧配了依赖，DataStream 侧 UDF 找不到包 → 混用就统一用 StreamExecutionEnvironment 配
• pipeline.classpaths 指向本地路径，集群节点根本访问不到 → 不确定就用 pipeline.jars / add_jars 上传
• 离线 cached_dir 的 wheel/源码包与集群平台/Python 版本不匹配 → 必须按集群环境构建缓存（Linux x86_64 + cp310/cp311）
• worker 端 Python 和 client 端 Python 版本不一致，Arrow/Pandas 相关依赖经常爆炸 → 统一版本，或者用 archive 自带解释器