使用Spark createDataFrame创建DataFrame时遇Python worker连接失败问题求助

问题背景 #

使用createDataFrame从本地列表创建DataFrame并调用take()时，触发Python worker failed to connect back错误，但通过spark.read.csv()读取外部文件完全正常。

报错代码 #

from pyspark.sql import SparkSession
# Create a Spark session
spark = SparkSession.builder\
        .appName("MyApp") \
        .getOrCreate()
person = spark.createDataFrame([
    (0, "AA", 0),
    (1, "BB", 1),
    (2, "CC", 1)
],schema= ["id", "name", "graduate"])
person.take(6)

核心错误信息 #

Py4JJavaError: An error occurred while calling o43.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0) (QUASAR executor driver): org.apache.spark.SparkException: Python worker failed to connect back.

可能的原因与解决办法 #

1. 强制Python环境一致性 #

Spark驱动端和Worker端必须使用完全相同的Python解释器，如果Worker启动时调用了不同版本的Python（比如系统默认Python vs 虚拟环境Python），就会出现连接失败。

• 解决：在初始化SparkSession时指定Python路径：

spark = SparkSession.builder\
        .appName("MyApp") \
        .config("spark.pyspark.python", "/path/to/your/python") \
        .config("spark.pyspark.driver.python", "/path/to/your/python") \
        .getOrCreate()

• 提示：可以用which python（Linux/macOS）或where python（Windows）查看当前使用的Python路径。

2. 调整Worker内存限制 #

创建内存中的DataFrame时，Python Worker需要足够内存处理数据，内存不足会导致连接中断。

• 解决：增加Worker的内存分配：

spark = SparkSession.builder\
        .appName("MyApp") \
        .config("spark.python.worker.memory", "2g") \
        .getOrCreate()

• 可根据实际情况调整内存值，比如1g、4g等。

3. 排查防火墙/安全软件拦截 #

本地防火墙或安全软件可能会拦截Spark驱动与Worker之间的本地网络通信，导致连接失败。

• 解决：临时关闭防火墙或安全软件测试，若恢复正常，添加Spark相关进程到白名单。

4. 检查版本兼容性 #

Spark与Python版本不兼容也会引发此类问题：

• Spark 3.x 兼容Python 3.7-3.11；Spark 2.x 兼容Python 2.7、3.4-3.7。
• 解决：升级或降级Python/Spark到匹配版本。

5. 清理Spark临时文件 #

Spark生成的临时Worker文件可能损坏，导致Worker启动失败。

• 解决：删除临时目录（Linux/macOS为/tmp/spark-*，Windows为C:\Users\<你的用户名>\AppData\Local\Temp\spark-*），重新运行代码。

验证代码 #

修改配置后，用以下代码验证是否修复：

from pyspark.sql import SparkSession

spark = SparkSession.builder\
        .appName("MyApp") \
        .config("spark.pyspark.python", "/path/to/your/python") \
        .getOrCreate()

person = spark.createDataFrame([
    (0, "AA", 0),
    (1, "BB", 1),
    (2, "CC", 1)
], schema=["id", "name", "graduate"])
print(person.take(6))

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