使用PyFlink Kafka Connector时Producer正常Consumer报找不到pyflink模块

1. 确认Python环境一致性 #

• 执行which python3获取当前使用的Python解释器路径，再执行pip3 show pyflink查看pyflink的安装位置（Location字段），确保两个路径属于同一环境。
• 如果使用了虚拟环境，必须先激活对应环境再执行读取脚本，比如source /path/to/venv/bin/activate。

2. 检查Flink Python配置 #

• 打开$FLINK_HOME/conf/flink-conf.yaml，确认python.executable配置指向正确的Python3路径，例如：

  python.executable: /usr/bin/python3
  

• 验证$FLINK_HOME/lib目录下是否存在pyflink相关的jar包（如pyflink-table-1.17.0.jar），若缺失可重新安装pyflink自动生成。

3. 调整脚本执行方式 #

• 优先使用Flink官方提供的pyflink run命令执行读取脚本，确保依赖环境正确加载：

  $FLINK_HOME/bin/pyflink run your_kafka_read_script.py
  

• 若坚持直接用python3执行，需在脚本开头手动添加pyflink路径到sys.path：

  import sys
  # 替换为pip3 show pyflink输出的Location路径
  sys.path.append("/usr/local/lib/python3.8/site-packages")
  from pyflink.table import EnvironmentSettings, TableEnvironment
  

4. 排查权限与缓存问题 #

• 检查pyflink安装目录的权限，确保执行用户有读取权限：

  chmod -R 755 $(pip3 show pyflink | grep Location | awk -F': ' '{print $2}')/pyflink
  

• 删除脚本目录下的__pycache__文件夹，或用python3 -B your_script.py执行以避免缓存干扰。

5. 重新验证pyflink安装 #

• 在当前Python环境中执行以下命令，确认pyflink可正常导入：

  python3 -c "import pyflink; print(pyflink.__version__)"
  

• 若报错，重新安装指定版本的pyflink：

  pip3 install pyflink==1.17.0 --force-reinstall --no-cache-dir
  

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