Spark 3.1.0大DataFrame高效合并及SparkContext关闭报错解决

在Spark 3.1.0环境下合并大型DataFrame时，遇到如下报错：
Py4JJavaError: An error occurred while calling o26434.collectToPython. : org.apache.spark.SparkException: Job 390 cancelled because SparkContext was shut down

当前采用循环调用union()的方式合并DataFrame，尝试替换union后仍出现相同错误，现寻求更高效的DataFrame合并方案。

def get_misc_exception(df: DataFrame, grouping_columns: list, distribution_cols: list) -> DataFrame:
    misc_df = None
    misc_cols_df_list = []
    
    for key, value in misc_cols_dict.items(): 
        print(f'{key}----------------')
        
        filtered_df = df.filter(value)
        
        if filtered_df.rdd.isEmpty():
            data = [(data_date, hour, key, 0, 0, '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')]
            misc_col_df = spark.createDataFrame(data, schema = final_df_schema)
            
        else:
            misc_col_df = get_exception(filtered_df, grouping_columns, distribution_cols)
            misc_col_df = misc_col_df.withColumn('Exception', F.lit(key)).select(*final_df_cols)
                
        misc_cols_df_list.append(misc_col_df)
        
    misc_df = misc_cols_df_list[0]
    
    for misc_cols_df in misc_cols_df_list[1:]:
        misc_df = misc_df.union(misc_cols_df)
    
    return misc_df

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<timed exec> in <module>

/usr/local/spark/python/pyspark/sql/pandas/conversion.py in toPandas(self)
    136 
    137         # Below is toPandas without Arrow optimization.
--> 138         pdf = pd.DataFrame.from_records(self.collect(), columns=self.columns)
    139         column_counter = Counter(self.columns)
    140 

/usr/local/spark/python/pyspark/sql/dataframe.py in collect(self)
    594         """
    595         with SCCallSiteSync(self._sc) as css:
--> 596             sock_info = self._jdf.collectToPython()
    597         return list(_load_from_socket(sock_info, BatchedSerializer(PickleSerializer())))
    598 

/usr/local/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1303         answer = self.gateway_client.send_command(command)
   1304         return_value = get_return_value(
-> 1305             answer, self.gateway_client, self.target_id, self.name)
   1306 
   1307         for temp_arg in temp_args:

/usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
    126     def deco(*a, **kw):
    127         try:
--> 128             return f(*a, **kw)
    129         except py4j.protocol.Py4JJavaError as e:
    130             converted = convert_exception(e.java_exception)

/usr/local/spark/python/lib/py4j-0.10.9-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}.
".
--> 328                     format(target_id, ".", name), value)
    329             else:
    330                 raise Py4JError(

Py4JJavaError: An error occurred while calling o26434.collectToPython.
: org.apache.spark.SparkException: Job 390 cancelled because SparkContext was shut down
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$cleanUpAfterSchedulerStop$1(DAGScheduler.scala:979)
    at org.apache.spark.scheduler.DAGScheduler.$anonfun$cleanUpAfterSchedulerStop$1$adapted(DAGScheduler.scala:977)
    at scala.collection.mutable.HashSet.foreach(HashSet.scala:79)
    at org.apache.spark.scheduler.DAGScheduler.cleanUpAfterSchedulerStop(DAGScheduler.scala:977)
    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onStop(DAGScheduler.scala:2257)
    at org.apache.spark.util.EventLoop.stop(EventLoop.scala:84)
    at org.apache.spark.scheduler.DAGScheduler.stop(DAGScheduler.scala:2170)
    at org.apache.spark.SparkContext.$anonfun$stop$12(SparkContext.scala:1973)
    at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1357)
    at org.apache.spark.SparkContext.stop(SparkContext.scala:1973)
    at org.apache.spark.SparkContext$$anon$3.run(SparkContext.scala:1922)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:775)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2120)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2139)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:2164)
    at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1004)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.rdd.RDD.withScope(RDD.scala:388)
    at org.apache.spark.rdd.RDD.collect(RDD.scala:1003)
    at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:385)
    at org.apache.spark.sql.Dataset.$anonfun$collectToPython$1(Dataset.scala:3450)
    at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3618)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:100)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:160)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:87)
    at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:764)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
    at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3616)
    at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:3447)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.base/java.lang.Thread.run(Thread.java:829)

1. 替换循环Union为批量合并 #

循环调用union会不断累加执行计划（Lineage），导致DAG过于复杂，最终引发资源耗尽或SparkContext异常关闭。改用functools.reduce批量合并：

from functools import reduce
from pyspark.sql import DataFrame

# 列顺序一致时用union
misc_df = reduce(DataFrame.union, misc_cols_df_list)

# 列顺序可能不一致时用unionByName（自动对齐列）
misc_df = reduce(lambda df1, df2: df1.unionByName(df2), misc_cols_df_list)

2. 优化空DataFrame判断逻辑 #

原代码中filtered_df.rdd.isEmpty()会触发额外的Action作业，改用Spark 2.4+支持的filtered_df.isEmpty()，直接在DataFrame层面判断，减少不必要的资源消耗：

if filtered_df.isEmpty():
    # 后续逻辑不变

3. 预创建空DataFrame复用 #

避免每次创建空DataFrame时都生成单行数据，可预先定义符合Schema的空DataFrame，减少重复操作：

# 函数外预定义
empty_misc_df = spark.createDataFrame([], schema=final_df_schema)

# 函数内判断为空时使用
if filtered_df.isEmpty():
    # 构造单行数据并合并到空DataFrame
    data = [(data_date, hour, key, 0, 0, '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}')]
    misc_col_df = empty_misc_df.union(spark.createDataFrame(data, schema=final_df_schema))

4. 排查SparkContext提前关闭的其他原因 #

• 检查代码中是否存在主动调用spark.stop()或sc.stop()的逻辑
• 确认集群资源配置是否足够，可尝试调高spark.driver.memory、spark.executor.memory参数
• 调整spark.sql.shuffle.partitions参数，避免过多分区导致资源耗尽

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