PySpark 3.2.1中SparkContext已停止异常的规避方法咨询

以下是触发异常的代码：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('jist-test').getOrCreate()
myDF = spark.read.parquet("v3io://projects/fs-demo-example/FeatureStore/vct_all_other_basic")
myDF.show(3)

运行后偶发如下异常：

Py4JJavaError                             Traceback (most recent call last)
<ipython-input-28-fd92fbf397a3> in <module>
      1 spark = SparkSession.builder.appName('jist-test').getOrCreate()
----> 2 myDF = spark.read.parquet("v3io://projects/fs-demo-example/FeatureStore/vct_all_other_basic")
/spark/python/pyspark/sql/readwriter.py in parquet(self, *paths, **options)
    299                        int96RebaseMode=int96RebaseMode)
    300 
---> 301         return self._df(self._jreader.parquet(_to_seq(self._spark._sc, paths)))
    302 
    303     def text(self, paths, wholetext=False, lineSep=None, pathGlobFilter=None,
/spark/python/lib/py4j-0.10.9.3-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1320         answer = self.gateway_client.send_command(command)
   1321         return_value = get_return_value(
-> 1322             answer, self.gateway_client, self.target_id, self.name)
   1323 
   1324         for temp_arg in temp_args:
/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
    109     def deco(*a, **kw):
    110         try:
---> 111             return f(*a, **kw)
    112         except py4j.protocol.Py4JJavaError as e:
    113             converted = convert_exception(e.java_exception)
/spark/python/lib/py4j-0.10.9.3-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    326                 raise Py4JJavaError(
    327                     "An error occurred while calling {0}{1}{2}.\n".
---> 328                     format(target_id, ".", name), value)
    329             else:
    330                 raise Py4JError(
Py4JJavaError: An error occurred while calling o91.parquet.
: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:
org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java
...

该异常在PySpark 3.2.1版本中偶尔出现，以下是几种可靠的规避方案：

1. 显式检查SparkContext状态，失效时重建Session #

在获取SparkSession前，先检查关联的SparkContext是否处于活跃状态，如果已停止则重新构建：

from pyspark.sql import SparkSession

def get_active_spark_session(app_name):
    spark = SparkSession.getActiveSession()
    if spark is None or spark.sparkContext._jsc.sc().isStopped():
        # 清理失效的Session/Context
        if spark is not None:
            spark.stop()
        # 重新创建Session
        spark = SparkSession.builder.appName(app_name).getOrCreate()
    return spark

spark = get_active_spark_session('jist-test')
myDF = spark.read.parquet("v3io://projects/fs-demo-example/FeatureStore/vct_all_other_basic")
myDF.show(3)

2. 避免重复创建/随意停止SparkSession #

• 在Notebook环境中，不要反复执行spark.stop()后立即调用getOrCreate()，这可能导致Context状态不一致。
• 整个应用生命周期内，优先复用已有活跃Session，确保创建逻辑幂等。

3. 调整SparkContext空闲超时配置 #

集群环境中（如YARN、K8s），SparkContext可能因空闲超时被资源管理器终止，可通过配置延长超时：

spark = SparkSession.builder.appName('jist-test') \
    .config("spark.network.timeout", "3600s") \
    .config("spark.executor.heartbeatInterval", "60s") \
    .getOrCreate()

根据集群实际情况调整参数，避免心跳超时导致Context意外停止。

4. 捕获异常并添加重试逻辑 #

针对偶发的停止异常，在数据读取等关键操作处添加重试机制：

from pyspark.sql import SparkSession
import time

def read_parquet_with_retry(spark, path, max_retries=3):
    for retry in range(max_retries):
        try:
            return spark.read.parquet(path)
        except Exception as e:
            if "Cannot call methods on a stopped SparkContext" in str(e):
                # 重建Session后重试
                spark.stop()
                spark = SparkSession.builder.appName('jist-test').getOrCreate()
                time.sleep(2)
                continue
            else:
                raise
    raise RuntimeError(f"重试{max_retries}次后仍无法读取Parquet文件")

spark = SparkSession.builder.appName('jist-test').getOrCreate()
myDF = read_parquet_with_retry(spark, "v3io://projects/fs-demo-example/FeatureStore/vct_all_other_basic")
myDF.show(3)

5. 确保资源释放逻辑正确 #

• 脚本环境中，不要中途随意停止SparkContext，应在所有任务完成后再调用spark.stop()。
• Notebook环境中，将Session创建逻辑放在单独的初始化单元格，后续单元格直接复用，避免重复定义。

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