【免费下载链接】flink
项目地址：
https://gitcode.com/gh_mirrors/fli/flink
点击查看 免费下载
本文围绕 PyFlink（Apache Flink 的 Python API）在真实开发中最常遇到的三类问题展开：如何准备并上传 Python 虚拟环境、如何为作业添加 Jar 与 Python 文件依赖，以及如何在 mini cluster（本地/IDE 内执行）中正确等待作业结束。结合 FAQ 官方文档、依赖管理文档 与 PyFlink 源码 中的实现细节，读完本文你将能独立完成 PyFlink 作业的依赖打包、提交与调试排错。
准备 Python 虚拟环境
PyFlink 作业中的 Python UDF 由独立的 Python worker 进程执行，因此集群节点上必须存在与作业匹配的 Python 环境。FAQ 官方推荐使用官方提供的一键脚本 setup-pyflink-virtual-env.sh 生成可随作业上传的虚拟环境压缩包，该脚本支持 Mac OS 和大多数 Linux 发行版。
脚本的执行逻辑如下（以当前仓库中的脚本内容为准）：
对应稳定版（stable）与不稳定版（unstable）文档，执行方式如下：
# stable 版本：指定 PyFlink 版本号
$ sh setup-pyflink-virtual-env.sh {{< version >}}
# 示例：sh setup-pyflink-virtual-env.sh 1.19.0
# unstable（开发版/未发布版本）：不传参数，安装最新 apache-flink
$ sh setup-pyflink-virtual-env.sh
提示：脚本内部依赖 wget、zip 等基础命令，执行前请确保本机已安装；生成产物为当前目录下的 venv.zip 与 venv/ 目录。
使用 Python 虚拟环境执行 PyFlink 作业
本地执行：先激活环境
在本地直接运行 PyFlink 脚本时，作业直接复用当前进程所在的 Python 环境，因此需要先激活刚才创建的虚拟环境：
# 激活 conda python 虚拟环境
$ source venv/bin/activate
$ python xxx.py
集群执行：上传归档并指定解释器
提交到集群时，需要把虚拟环境作为 archive 上传，并显式指定 Python UDF worker 使用的解释器路径。FAQ 给出的最小配置如下：
# 指定 Python 虚拟环境归档文件
table_env.add_python_archive("venv.zip")
# 指定用于执行 Python UDF worker 的 Python 解释器路径
table_env.get_config().set_python_executable("venv.zip/venv/bin/python")
从源码看，这两条调用都发生在 Python Table API 的 TableEnvironment 上：
- table_environment.py#L1257 中的 add_python_archive(archive_path, target_dir=None) 会将归档路径（若指定了 target_dir，则以 # 分隔符拼接）写入 PythonOptions.PYTHON_ARCHIVES 配置项，执行时归档文件会被解压到 Python UDF worker 的工作目录；
- set_python_executable 将解释器路径写入 PythonOptions.PYTHON_EXECUTABLE 配置项，用于启动 Python worker 进程。
数据流 API 的对应方法是 StreamExecutionEnvironment.add_python_archive 与 set_python_executable，实现在 stream_execution_environment.py#L658 与 stream_execution_environment.py#L721。需要注意：
- 虚拟环境必须与集群所在平台（操作系统、架构）以及集群要求的 Python 版本匹配；
- 归档文件目前仅支持 zip 格式（zip、jar、whl、egg 等）以及 tar 系列格式（tar、tar.gz、tgz），不支持 7z、rar 等格式；
- add_python_archive 也常用于在 UDF 内以相对路径访问数据文件（如读取模型文件），例如 add_python_archive("/path/to/py_env.zip", "myenv") 后，在 UDF 中可用 open("myenv/py_env/data/data.txt") 访问归档内文件；
- 如果解释器路径指向归档文件内部，应使用相对路径而非绝对路径。
配置方式与命令行方式
除 API 外，上述能力都可以通过配置项或提交命令等价实现，详见 Python 配置文档 与 CLI 使用文档：
| Python 文件/包/目录 | add_python_file | python.files | -pyfs, –pyFiles |
| requirements.txt | set_python_requirements | python.requirements | -pyreq, –pyRequirements |
| 归档文件 | add_python_archive | python.archives | -pyarch, –pyArchives |
| UDF worker 解释器 | set_python_executable | python.executable | -pyexec, –pyExecutable |
| 客户端解释器 | 激活当前 shell 环境 | python.client.executable | -pyclientexec, –pyClientExecutable |
CLI 提交示例（参考 cli.md#L501）：
flink run \\
-pyarch shipfiles/venv.zip \\
-pyclientexec venv.zip/venv/bin/python3 \\
-pyexec venv.zip/venv/bin/python3 \\
-pyfs shipfiles \\
-pyreq requirements.txt \\
-c org.apache.flink.python.PythonFunctionRunner python_job.py
其中客户端解释器（python.client.executable）用于在客户端解析 Python UDF 定义，也可以通过环境变量 PYFLINK_CLIENT_EXECUTABLE 指定；而 python.executable 决定的是集群上执行 UDF 的 worker 解释器，两者职责不同，都不可省略。
为 PyFlink 作业添加 Jar 文件
PyFlink 作业可能依赖连接器（connector）、Java UDF 等 Jar 包。FAQ 提供了两种基于配置项的添加方式：
# 注意：仅支持本地文件 URL（以 "file://" 开头），多个 Jar 用 ";" 分隔
# pipeline.jars 指定的 Jar 会被上传到集群
table_env.get_config().set("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/udf.jar")
# 注意：路径必须指定协议（如 file://），且需保证这些 URL 在客户端与集群上都能访问
# pipeline.classpaths 指定的 Jar 会加入作业执行的 classpath
table_env.get_config().set("pipeline.classpaths", "file:///my/jar/path/connector.jar;file:///my/jar/path/udf.jar")
- pipeline.jars：Jar 会被上传到集群并随作业分发，适合本地构建、需跨节点传递的 Jar；
- pipeline.classpaths：仅把 URL 加入客户端与集群的 classpath，要求 URL 在两端都可访问（如已放置在共享文件系统上），不会上传文件。
Python DataStream API 提供了语义相同的方法 add_jars / add_classpaths，底层分别写入 PipelineOptions.JARS 与 PipelineOptions.CLASSPATHS 配置：
stream_execution_environment.add_jars("file:///my/jar/path/connector1.jar", "file:///my/jar/path/connector2.jar")
stream_execution_environment.add_classpaths("file:///my/jar/path/connector1.jar")
也可以直接在提交作业时使用命令行参数 –jarfile <jarFile> 指定。注意：–jarfile 一次只能指定一个 Jar 文件，如果存在多个依赖 Jar，需要预先打成 fat jar（参见 依赖管理文档）。
为 PyFlink 作业添加 Python 文件
当 UDF 依赖自定义 Python 模块、包或本地目录时，可以使用命令行参数 -pyfs/–pyFiles，或在代码中调用 TableEnvironment.add_python_file。这些依赖会被加入 Python UDF worker 的 PYTHONPATH，从而可直接 import。
假设目录结构如下：
myDir
├──utils
├──__init__.py
├──my_util.py
按 FAQ 中的示例添加并导入：
table_env.add_python_file('myDir')
def my_udf():
from utils import my_util
…
从 table_environment.py#L1195 的实现可见，add_python_file 会把路径累积写入 python.files 配置项（多个路径以文件分隔符拼接）；DataStream API 的对应方法 add_python_file 实现在 stream_execution_environment.py#L598，逻辑一致。
如果 UDF 还依赖第三方 Python 库，推荐使用 requirements.txt 方式，配合 set_python_requirements 在集群上离线安装：
# 生成 requirements.txt（可手动精简）
$ echo numpy==1.16.5 >> requirements.txt
$ echo pandas==1.0.0 >> requirements.txt
# 或直接导出当前环境全部依赖
$ pip freeze > requirements.txt
# 为无法在集群访问的依赖准备离线安装包目录（可选）
$ pip download -d cached_dir -r requirements.txt –no-binary :all:
# requirements_cache_dir 为可选参数，用于离线安装
table_env.set_python_requirements(
requirements_file_path="/path/to/requirements.txt",
requirements_cache_dir="cached_dir")
两点约束值得注意：
- 离线安装包必须与集群平台和 Python 版本匹配；
- 集群安装依赖依赖 pip（版本 >= 20.3）与 setuptools（版本 >= 37.0.0）。
在 mini cluster 中执行作业：记得等待异步任务结束
当在 mini cluster（例如在 IDE 中本地运行）执行 PyFlink 作业时，以下 API 是异步的：Python Table API 的 TableEnvironment.execute_sql、StatementSet.execute 等，以及 Python DataStream API 的 StreamExecutionEnvironment.execute_async。如果程序在主线程直接退出，作业可能尚未执行完，导致看不到执行结果。FAQ 给出的正确写法是显式等待：
# 异步执行 SQL / Table API 查询，然后等待完成
t_result = table_env.execute_sql(…)
t_result.wait()
# 异步执行 DataStream 作业，然后等待完成
job_client = stream_execution_env.execute_async('My DataStream Job')
job_client.get_job_execution_result().result()
从源码看，execute_async 在 stream_execution_environment.py#L826 中返回一个 JobClient 对象，作业提交后立即返回，真正的执行结果需要通过对该客户端调用 get_job_execution_result().result() 阻塞获取——这也正是 FAQ 中 wait() / result() 调用背后的机制。
注意：在远程集群上执行作业时不需要等待，提交后程序即可返回。因此上述等待代码只应在本地/IDE 调试场景中使用，提交到集群前记得移除，避免不必要的阻塞。
总结
PyFlink 作业从本地调试到集群上线的常见坑点可以归纳为四步排查：
FAQ 原文、依赖管理文档、Python 配置文档 与 CLI 文档 分别覆盖了上述能力的详细说明，本文涉及的 API 实现均可在 flink-python/pyflink/table/table_environment.py 与 flink-python/pyflink/datastream/stream_execution_environment.py 中进一步追溯验证。
赞
【免费下载链接】flink
项目地址：
https://gitcode.com/gh_mirrors/fli/flink
点击查看 免费下载
相关推荐
Cyber Engine Tweaks 终极指南：为什么这个工具能让你的赛博朋克2077体验翻倍？🔥
Tiny C Compiler终极指南：如何用最小的编译器获得最快的开发体验
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
