Spark任务间超大时间间隔，查询优化耗时占90%求助排查

我开发了一个Spark应用，通过withColumn为DataFrame添加500+个复杂且相互关联的计算列，逐个定义并添加这些列。所有代码写完后，执行df.persist().count()触发全量计算，之后写入Hive表。

遇到的问题：

• 读取数据并完成500个列的定义（到执行df.count()前）仅需10-15分钟
• 之后Spark会进入1.5-2小时的无响应等待，Spark UI没有任何相关日志
• 等待结束后，count计算和写入表仅需10-15分钟

我曾尝试用selectExpr替代withColumn，但因为500个列相互关联，会出现列名歧义错误，无法实现。

示例伪代码：

# 添加abc列所需的7个属性
df = calling7function().retrieve(self.spark, self.context_dict)

# 第一次定义abc列
df = df.withColumn("abc",
    F.when((df.x=='Y'), 'OS') \
    .when((df.t=='Y'), 'SR') \
    .when((df.a=='Y'), '3F') \
    .when((df.v=='Y'), 'SS') \
    .when((df.w=='Y'), 'DL') \
    .when((df.v=='Y'), 'RO') \
    .when((df.wq=='Y'), 'RS') \
    .when((df.as=='Y'), 'CO') \
    .otherwise('')
)
# 更新abc列
df = df.withColumn("abc",
    F.when((df.asd.isNotNull()) & (df.abc=='') & (df.rsx.isin(['6','7','9'])), 'RS') \
    .when((df.asd.isNotNull()) & (df.abc=='') & (df.rsx=='1'), 'CO') \
    .when((df.asd.isNotNull()) & (df.abc=='') & (df.rsx=='13'), 'SS') \
    .when((df.asd.isNotNull()) & (df.abc=='') & (df.rsx=='14'), '3F').otherwise(df.abc))

# 再次更新abc列
df = df.withColumn("abc",F.when((df.ax=='qq') & (df.aq=='OOO'), 'RO').when((df.ae=='we') & (df.qwe=='OOO'), 'RS').otherwise(df.abc))

# 最后一次更新abc列
df = df.withColumn("abc",F.when((df.abc=='') , 'PO').otherwise(df.abc))

一、等待时间内发生了什么？ #

这段等待是Spark的查询计划优化阶段，核心原因如下：

1. 逻辑计划膨胀：500+个withColumn调用（尤其是多次更新同一列）会生成极其复杂的逻辑执行计划树。Spark需要遍历这棵庞大的树，处理大量列依赖、重复计算节点，进行规则匹配和优化。
2. 物理计划生成阻塞：复杂逻辑计划转换为物理计划时，Catalyst优化器要完成大量耗时操作：
   • 解析所有列的依赖关系，消除冗余计算
   • 合并多次更新同一列产生的重复节点
   • 对500+列的表达式做类型检查、常量折叠等优化
3. UI日志延迟：优化阶段属于Driver端本地计算，不会生成Task级日志，因此Spark UI无相关输出，直到物理计划生成完成后才会提交任务。

二、排查方法 #

• 开启Catalyst日志：在Driver的log4j配置中添加log4j.logger.org.apache.spark.sql.catalyst=DEBUG，查看优化阶段的详细步骤，定位耗时最长的优化环节。
• 导出执行计划：在df.count()前执行df.explain(mode="extended")，将计划保存到文件，分析是否存在大量重复的Project节点或异常复杂的依赖链。
• 监控Driver资源：检查Driver的CPU、内存使用情况，优化阶段会占用大量CPU，内存不足还会触发频繁GC，进一步拖慢速度。

三、修复/缩短时长的方案 #

1. 合并列更新逻辑，减少withColumn调用 #

多次用withColumn更新同一列会生成大量冗余Project节点，可将所有更新逻辑合并到一次调用中：

# 合并所有abc列的逻辑到一次withColumn调用
df = df.withColumn("abc",
    F.when((df.x=='Y'), 'OS') \
    .when((df.t=='Y'), 'SR') \
    .when((df.a=='Y'), '3F') \
    .when((df.v=='Y'), 'SS') \
    .when((df.w=='Y'), 'DL') \
    .when((df.v=='Y'), 'RO') \
    .when((df.wq=='Y'), 'RS') \
    .when((df.as=='Y'), 'CO') \
    # 追加后续更新逻辑
    .when((df.asd.isNotNull()) & (F.col('abc')=='') & (df.rsx.isin(['6','7','9'])), 'RS') \
    .when((df.asd.isNotNull()) & (F.col('abc')=='') & (df.rsx=='1'), 'CO') \
    .when((df.asd.isNotNull()) & (F.col('abc')=='') & (df.rsx=='13'), 'SS') \
    .when((df.asd.isNotNull()) & (F.col('abc')=='') & (df.rsx=='14'), '3F') \
    .when((df.ax=='qq') & (df.aq=='OOO'), 'RO') \
    .when((df.ae=='we') & (df.qwe=='OOO'), 'RS') \
    .when((F.col('abc')==''), 'PO') \
    .otherwise('')
)

这种方式能大幅压缩逻辑计划的节点数量，降低优化器的处理负担。

2. 用select替代重复withColumn，显式控制列集合 #

不要逐个withColumn添加列，而是一次性在select中定义所有新列，同时保留原有列：

# 一次性定义所有需要的新列，避免多次Project操作
df = df.select(
    "*",  # 保留原有列
    # 定义第一个计算列
    F.when(...).alias("col1"),
    # 基于col1定义col2
    F.when(F.col("col1") == ..., ...).alias("col2"),
    # 其他500个列依次定义
    ...
)

生成的逻辑计划更紧凑，优化器处理效率更高。

3. 调整Spark优化器配置 #

• 关闭不必要的优化规则：通过spark.sql.optimizer.excludedRules排除耗时过长的规则，比如不需要列裁剪时可排除org.apache.spark.sql.catalyst.optimizer.ColumnPruning。
• 增大Driver内存：优化阶段需要处理大量计划节点，给Driver分配足够内存（如--driver-memory 16g），减少GC停顿。
• 提高Catalyst并行度：设置spark.sql.catalyst.parallelism（默认8），若Driver CPU核心充足，适当调高该值，让优化器并行处理计划节点。

4. 分阶段持久化，拆分计算逻辑 #

把500个列分成若干组，每组计算完成后立即persist并触发计算，再基于持久化后的DataFrame计算下一组列：

# 第一组：计算200个列
df_stage1 = df.withColumn(...).withColumn(...)
df_stage1.persist().count()

# 第二组：基于stage1计算200个列
df_stage2 = df_stage1.withColumn(...).withColumn(...)
df_stage2.persist().count()

# 第三组：完成剩余100个列
df_final = df_stage2.withColumn(...).withColumn(...)
df_final.persist().count()

每次优化的计划复杂度降低，避免一次性处理500个列的庞大计划。

5. 规避selectExpr的列名歧义 #

对于selectExpr的歧义问题，可给中间列起临时别名，计算完成后再删除：

df = df.selectExpr(
    "*",
    "CASE WHEN x='Y' THEN 'OS' ... ELSE '' END AS abc_temp",
    "CASE WHEN asd IS NOT NULL AND abc_temp='' AND rsx IN ('6','7','9') THEN 'RS' ... ELSE abc_temp END AS abc"
).drop("abc_temp")

虽然写法繁琐，但能避免列名歧义，同时减少withColumn带来的计划膨胀。

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