Spark+Hadoop+PySpark环境下Spark Context未初始化报错求助

错误日志 #

ApplicationMaster: Waiting for spark context initialization...
25/05/12 23:56:11 INFO ApplicationMaster: Final app status: FAILED, exitCode: 13
25/05/12 23:56:11 ERROR ApplicationMaster: Uncaught exception: java.lang.IllegalStateException: User did not initialize spark context!

相关代码 #

主程序sPCA.py #

import pyspark
import numpy as np
from pyspark.sql import SparkSession
from utils import YtXSparkJob
import random


def main(args):
    # inicializar spark
    spark = SparkSession.builder.appName("sPCA").getOrCreate()
    sc = spark.sparkContext  # así obtienes el SparkContext moderno
    random.seed(42)
    np.random.seed(42)
    # ... 其他业务代码
    YtXSparkJob(..., spark)  # 传入SparkSession实例


if __name__ == '__main__':
    """Run sPCA"""
    # ... 参数解析等代码
    main(args)

utils.py相关函数 #

def YtXSparkJob(Y, Ym, Xm, CM, D, d, spark):
    sc = spark.sparkContext
    # 需要使用sc广播变量等操作

任务提交命令 #

spark-submit --master yarn --deploy-mode cluster --archives env.tar.gz#environment --py-files accumulators.py utils.py sPCA.py --input "hdfs:///grupoh/challenge/input/datos_1.txt" --dim 2 --maxIters 10 --output "hdfs:///grupoh/challenge/output/datos_1_spark_pca.txt"

排查情况 #

通过Spark集群节点UI确认，错误源自utils.py中获取Spark Context的操作，当前通过传入SparkSession实例来获取sc，但初始化失败。

解决方案 #

方案1：优先完成SparkSession初始化 #

将SparkSession的初始化放在main函数最开头，避免在初始化前执行耗时/阻塞操作，确保上下文优先建立：

def main(args):
    # 优先初始化SparkSession
    spark = SparkSession.builder.appName("sPCA").getOrCreate()
    sc = spark.sparkContext
    # 再执行其他初始化逻辑
    random.seed(42)
    np.random.seed(42)
    # ... 后续业务代码

方案2：直接传入SparkContext到工具函数 #

简化依赖链，直接将已初始化的SparkContext传入工具函数，避免二次获取：

# sPCA.py中调用时传入sc
YtXSparkJob(..., sc)

# utils.py修改函数定义
def YtXSparkJob(Y, Ym, Xm, CM, D, d, sc):
    # 直接使用sc操作
    broadcast_var = sc.broadcast(...)

方案3：检查集群环境依赖 #

集群模式下需确保环境与依赖匹配：

• 确认env.tar.gz中的Python环境PySpark版本与集群Spark版本完全一致，避免版本兼容问题
• 验证--py-files参数包含的utils.py、accumulators.py路径正确，集群节点能正确加载这些文件

方案4：添加初始化有效性检查 #

在工具函数中增加检查逻辑，提前发现无效上下文：

def YtXSparkJob(Y, Ym, Xm, CM, D, d, spark):
    sc = spark.sparkContext
    if not sc or sc._jsc is None:
        raise RuntimeError("Spark Context未成功初始化，请检查会话建立逻辑")
    # 后续业务操作

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