PySpark遇java.io.EOFException及内存耗尽问题求解决

可能的原因 #

• Python Worker内存配额不足：默认PySpark给Python Worker分配的内存有限，即便看似简单的DataFrame，若列拆分操作引发隐式数据膨胀（比如拆分后爆炸生成大量行），会直接耗尽Worker内存导致进程崩溃，Java端因无法获取Worker响应触发java.io.EOFException。
• 列拆分逻辑的隐式数据膨胀：使用split+explode组合时，若原数据存在包含大量分隔符的行，会瞬间生成数万甚至数十万行数据，超出Worker内存承载上限。
• Spark环境内存配置过低：Driver或Executor的内存配置不足，导致Java端无法将数据传递给Python Worker，间接引发Worker崩溃。
• 环境兼容性问题：Python版本与PySpark版本不匹配、py4j依赖版本不一致，会导致Worker进程启动或运行时异常崩溃。

解决办法 #

• 调整Python Worker内存：
  • 设置环境变量：export PYSPARK_PYTHON_WORKER_MEMORY=4g（根据机器内存调整，建议不超过物理内存的1/3）。
  • 或在SparkSession初始化时配置：

    from pyspark.sql import SparkSession
    spark = SparkSession.builder \
        .config("spark.executor.pyspark.memory", "4g") \
        .getOrCreate()
    

• 优化列拆分逻辑：
  • 先通过抽样排查数据：df.sample(0.05).count()，确认是否是特定数据行引发的问题。
  • 避免无限制拆分：拆分后仅保留必要元素，比如split(col("col_name"), ",").getItem(0)，或过滤拆分后元素过多的行：filter(size(split(col("col_name"), ",")) < 100)。
  • 优先使用Spark内置函数：避免自定义UDF，内置split、explode等函数在Java端执行，不占用Python Worker内存。
• 提升Spark内存配置：
  • 调整Driver内存：spark.driver.memory=4g。
  • 调整Executor内存：spark.executor.memory=8g、spark.executor.cores=4（根据集群资源调整）。
• 校验环境兼容性：
  • 确认Python版本与PySpark版本匹配（如PySpark 3.4支持Python 3.8-3.11）。
  • 检查py4j版本：确保py4j版本与PySpark自带版本一致（可通过pip show py4j查看，若不一致则卸载后重新安装对应版本）。
• 排查异常数据：
  • 扫描数据中的超长文本行：df.filter(length(col("col_name")) > 10000).show()，对这类行做预处理（截断或过滤）后再执行拆分操作。

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