Windows环境下PyFlink与Kafka数据读取失败问题求助

这个报错的核心原因是Windows环境下，PyFlink的Python UDF执行环节依赖的临时资源打包、存储逻辑出现异常——stage bundle factory负责将Python代码打包分发给Worker节点，Windows的路径权限、默认临时目录配置、路径分隔符都可能触发这个问题。同时消费者offset无变化、lag持续增长，说明消费端根本没正常拉取到数据，和这个创建失败的问题直接相关。

• 显式配置Python环境与临时目录
  在代码开头添加配置，指定无空格、非中文的Python解释器路径和临时目录（手动提前创建好目录），避免Windows默认路径的权限或转义问题：

  from pyflink.datastream import StreamExecutionEnvironment
  env = StreamExecutionEnvironment.get_execution_environment()
  # 替换成你的Python实际安装路径
  env.set_python_executable("C:/Python39/python.exe")
  # 手动创建该目录，确保有读写权限
  env.set_python_temp_dir("D:/pyflink_temp")
  

• 确保依赖版本匹配
  检查PyFlink版本与Kafka连接器版本一致（比如PyFlink 1.15.x对应flink-connector-kafka-1.15.x_2.12.jar），将对应jar包放入Flink安装目录的lib文件夹，重启本地Flink集群（如果使用集群模式）。

• 重置消费者offset并调整并行度
  测试阶段先简化配置，避免并行度或残留offset干扰：

  1. 重置消费者组test_group_1的offset到最早位置（进入Kafka的bin目录执行）：

     kafka-consumer-groups.bat --bootstrap-server localhost:9092 --group test_group_1 --reset-offsets --to-earliest --all-topics --execute
     

  2. 在PyFlink代码中设置消费流并行度为1，并关闭checkpoint：

     # 假设kafka_source是你的Kafka消费数据源
     kafka_source = kafka_source.set_parallelism(1)
     env.disable_checkpointing()
     

• 临时关闭Windows安全软件
  Windows Defender或第三方杀毒软件可能拦截PyFlink创建临时文件、启动Python Worker进程，临时关闭实时保护后重新测试，确认是否是权限拦截导致的问题。

1. 执行offset重置命令后，重新运行PyFlink程序
2. 查看Kafka Tool中test_group_1的offset是否开始递增，lag是否逐步降低
3. 检查程序控制台是否输出预期的处理结果

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