PyFlink添加外部依赖至UDF时依赖找不到的问题排查

为什么依赖找不到？ #

你直接将依赖安装在当前根目录，打包时虽包含了依赖文件，但PyFlink默认不会自动将这些依赖路径加入Python的sys.path，导致运行时无法定位模块。

修正步骤 #

1. 规范依赖安装路径 #

不要把依赖直接装在当前根目录，创建单独的lib文件夹存放依赖：

pip install -r requirements.txt --target=lib

后续集成boto3时，只需把boto3添加到requirements.txt，重新执行上述命令即可。

2. 正确打包代码与依赖 #

打包时明确包含业务代码和lib目录，避免混入__pycache__、requirements.txt等不必要文件：

# 假设你的UDF代码文件名为udf_functions.py
zip -r pyflink.zip udf_functions.py lib/

3. 提交作业时指定依赖 #

提交Flink作业时，必须通过--py-files参数指定打包好的zip文件，让Flink将依赖路径加入Python运行环境：

flink run \
  --python udf_functions.py \
  --py-files pyflink.zip

集群模式下也可使用--py-archive参数指定归档文件，确保依赖被正确解压到可访问路径：

flink run \
  --python udf_functions.py \
  --py-archive pyflink.zip::/opt/flink/deps

4. 补全UDF代码的必要修正 #

你的UDF代码存在两处问题，需要补充导入并修正逻辑：

import requests
import json
import logging
from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())
def get_data():
    response = requests.get("https://api_endpoint")  # 补全URL的//，否则请求会失败
    logging.info(response.status_code)  # 建议打印状态码而非整个response对象，更实用
    return json.dumps(response.json())  # 注意：UDF声明返回STRING，需将字典转成JSON字符串返回

额外注意事项 #

• 确保本地Python版本与Flink集群的Python版本一致，避免依赖兼容性问题
• 如果依赖包含C扩展（比如boto3的部分依赖），需保证集群环境与本地编译环境一致，或使用纯Python版本的依赖包

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