PySpark多SparkSession创建致主程序写入操作死锁问题求助

核心问题根源 #

PySpark的SparkSession.builder.getOrCreate()虽会复用已有实例，但如果在不同模块中重复对SparkContext做配置（比如setCheckpointDir、addPyFile），会导致JVM内部状态冲突。同时，多Session实例操作Hive表和Parquet写入时，会触发资源锁竞争，最终引发死锁。

解决方案 #

1. 全局单例管理SparkSession #

创建独立的SparkSession管理模块，确保全应用只有一个实例，配置仅初始化一次：

# src/spark_session.py
import os
from pyspark.sql import SparkSession

# 全局单例变量
_global_spark = None

def get_spark():
    global _global_spark
    if _global_spark is None:
        # 初始化时一次性完成所有配置
        _global_spark = SparkSession.builder \
            .appName("X") \
            .enableHiveSupport()  # 必须启用Hive支持，否则无法操作Hive表
            .getOrCreate()
        
        # 仅在第一次初始化时设置上下文参数
        _global_spark.sparkContext.setCheckpointDir(
            os.getenv("CHECKPOINTS_DIR", "/tmp/checkpoints")
        )
        _global_spark.sparkContext.addPyFile("src.zip")
    return _global_spark

所有模块统一导入使用：

# 主程序和其他文件中
from spark_session import get_spark

spark = get_spark()
# 后续所有操作都使用这个spark实例

2. 清理重复配置与Session创建代码 #

• 删除其他文件中单独创建SparkSession的代码，替换为上述单例方法获取实例。
• 检查是否有重复执行setCheckpointDir、addPyFile的逻辑，这些操作对同一个SparkContext重复执行会破坏内部状态。

3. 修复写入代码的冗余参数 #

你的写入代码中重复指定了mode="overwrite"，属于冗余配置，去掉其中一个：

# 修正后的写入代码
self.spark.table(self.table_to_write).write.mode("overwrite").parquet(self.destination)

4. 临时应急排查方案 #

如果暂时无法重构代码，可在写入前重置SparkContext状态：

# 取消所有未完成的Job，释放资源
self.spark.sparkContext.cancelAllJobs()
# 强制刷新Session状态
self.spark.catalog.clearCache()
# 执行写入
self.spark.table(self.table_to_write).write.mode("overwrite").parquet(self.destination)

关键注意事项 #

• PySpark应用中必须保证只有一个SparkSession/SparkContext实例，多实例会引发JVM资源竞争、元数据不一致等问题。
• 操作Hive表时，初始化SparkSession必须添加.enableHiveSupport()，否则无法读取Hive元数据。

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