PySpark含聚合操作的大嵌套表任务执行失败的排查与优化咨询

看起来你遇到了PySpark处理大嵌套表聚合时的资源瓶颈和执行失败问题，结合你给出的场景和尝试过的操作，我来帮你梳理排查方向和优化方案，既能适配当前100M级别数据，也能为未来数十亿行的场景铺路：

一、先定位核心问题：Executor死亡+Shuffle失败的可能原因 #

从报错ExecutorDeadException和FetchFailedException来看，大概率是这几个原因叠加导致的：

1. Shuffle阶段资源耗尽：你的聚合操作涉及大量collect_set、countDistinct这类需要Shuffle的算子，嵌套列本身序列化/反序列化开销远高于扁平表，加上100M行的数据量，Shuffle时的磁盘/内存压力直接打挂Executor
2. 数据倾斜隐患：你提到最大分组有68K行，虽然单看数值不算极端，但如果colA/colB/colC/colD分组键存在少数高频值，会导致单个Task处理远超过平均值的数据量，OOM后引发Executor死亡
3. 嵌套列的额外开销：原始嵌套表的元数据冗余多，Spark处理时需要频繁解析嵌套结构，这也是你扁平化后Parquet体积从11.4G降到4.4G的核心原因——冗余元数据被清理了

二、针对性优化方案（按优先级排序） #

1. 先修正聚合逻辑的低级错误 #

你的聚合代码里有个明显的别名冲突问题：同一个colE被同时用在collect_set和sum的别名上，这不仅会导致逻辑错误，还可能让Spark执行计划混乱，先修正这个：

# 修正别名冲突的聚合代码
from pyspark.sql.functions import count, sort_array, collect_set, sum, min, max, countDistinct, first

df_agg = df_1.groupby("colA","colB","colC","colD").agg(
    count("*").alias("numRecords"),
    # 给colE的不同聚合结果单独命名，避免覆盖
    sort_array(collect_set("colE")).alias("colE_distinct_sorted"),
    sum("colE").alias("colE_total"),
    sum("colF").alias("colF_total"),
    sum("colG").alias("colG_total"),
    sum("colH").alias("colH_total"),
    min("colI").alias("colI_min"),
    max("colJ").alias("colJ_max"),
    countDistinct("colK").alias("colK_distinct_count"),
    first("colL").alias("colL_first"),
    first("colM").alias("colM_first"),
    first("colN").alias("colN_first"),
    first("colO").alias("colO_first"),
    sort_array(collect_set("colP")).alias("colP_distinct_sorted"),
    sort_array(collect_set("colQ")).alias("colQ_distinct_sorted"),
    max("colR").alias("colR_max"),
    max("colS").alias("colS_max")
)

2. 极致扁平化+提前过滤，从源头减负载 #

你已经尝试过扁平化，可以再优化两步：

• 扁平化时只保留聚合需要的字段，避免冗余数据占用内存
• 提前计算newCol并过滤无效行，减少后续聚合的数据量

from pyspark.sql.functions import when, col

# 一次性完成扁平化、newCol计算、无效数据过滤
df_flat = df.select(
    # 分组用的字段直接提取（如果是嵌套结构就用col("colA.field")）
    col("colA").alias("colA"),
    col("colB").alias("colB"),
    col("colC").alias("colC"),
    col("colD").alias("colD"),
    # 提前计算newCol，避免聚合阶段重复计算
    when((col("colA.field") == 1) & (col("colB.field1") == 2), col("colA.field1"))
    .otherwise(col("colB.field1")).alias("newCol"),
    # 聚合需要的其他字段
    col("colE"), col("colF"), col("colG"), col("colH"),
    col("colI"), col("colJ"), col("colK"),
    col("colL"), col("colM"), col("colN"), col("colO"),
    col("colP"), col("colQ"), col("colR"), col("colS")
).filter(
    # 提前过滤掉colE/colF等聚合字段为null的行
    col("colE").isNotNull() & col("colF").isNotNull()
)

3. 调整Spark配置，给Shuffle“松绑” #

针对Executor死亡和Shuffle失败，修改核心配置缓解资源压力：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("LargeAggregationJob") \
    # 增大Executor内存，根据集群资源调整，比如16G
    .config("spark.executor.memory", "16g") \
    # 堆外内存给足，处理Shuffle溢出
    .config("spark.executor.memoryOverhead", "4g") \
    # 调整Shuffle分区数，100M行建议设为1500-2000，让每个Task处理60K-70K行
    .config("spark.sql.shuffle.partitions", "1500") \
    # 开启自适应Shuffle，自动调整分区数
    .config("spark.sql.adaptive.enabled", "true") \
    # 开启BloomFilter优化countDistinct性能
    .config("spark.sql.optimizer.runtime.bloomFilter.enabled", "true") \
    # 允许Shuffle数据溢出到磁盘，避免直接OOM
    .config("spark.shuffle.spill.enabled", "true") \
    .getOrCreate()

4. 数据倾斜的针对性优化 #

如果调整配置后还是失败，就要处理数据倾斜：

方案A：加盐打散高频分组键

如果某个分组键（比如colA的某个值）是高频值，给它加随机前缀打散，聚合后再合并：

from pyspark.sql.functions import rand, floor, concat, lit, split, explode

# 1. 给高频分组键加盐（假设"高频值"是colA的倾斜键）
df_salted = df_flat.withColumn(
    "colA_salted",
    when(col("colA") == "高频值", concat(col("colA"), lit("_"), floor(rand()*10)))
    .otherwise(col("colA"))
)

# 2. 用加盐后的键做第一次聚合
df_agg_salted = df_salted.groupby("colA_salted","colB","colC","colD").agg(
    count("*").alias("numRecords"),
    sort_array(collect_set("colE")).alias("colE_distinct_sorted"),
    sum("colE").alias("colE_total"),
    # 其他聚合项和之前一致
)

# 3. 去掉盐值，合并高频分组的结果
df_agg_final = df_agg_salted.withColumn(
    "colA",
    when(col("colA_salted").contains("_"), split(col("colA_salted"), "_")[0])
    .otherwise(col("colA_salted"))
).groupby("colA","colB","colC","colD").agg(
    sum("numRecords").alias("numRecords"),
    # 合并去重后的colE集合
    sort_array(collect_set(explode(col("colE_distinct_sorted")))).alias("colE_distinct_sorted"),
    sum("colE_total").alias("colE_total"),
    # 其他聚合项：sum的直接加和，min/max取对应值
)

方案B：用近似去重替代精确去重

如果业务允许少量误差，把countDistinct换成approx_count_distinct，性能提升数倍：

from pyspark.sql.functions import approx_count_distinct

df_agg = df_flat.groupby("colA","colB","colC","colD").agg(
    # ... 其他聚合项
    approx_count_distinct("colK", rsd=0.02).alias("colK_distinct_count"),
    # rsd是相对误差，默认0.05，越小越精确但性能越差
    # ...
)

5. 为未来数十亿行的场景做前置优化 #

如果要支撑数十亿行的数据量，必须做架构级优化：

1. 分区分桶存储：把扁平化后的表按colA/colB分区，按colC/colD分桶，聚合时直接利用分区剪枝和分桶预排序，减少Shuffle量
2. 增量聚合：用Delta Lake或Hudi做增量计算，每次只处理新增数据，避免全量扫描
3. 离线预处理：提前把嵌套列扁平化成分区分桶表，后续聚合直接读预处理后的表，省去实时解析嵌套列的开销

三、验证步骤（从小到大，逐步排查） #

1. 先在扁平化后的4.4G表上跑简化聚合：只做count(*)和sum(colE)，验证基本聚合是否能成功
2. 逐步添加聚合项，比如先加collect_set，再加countDistinct，定位哪个算子导致失败
3. 用spark.sql("SELECT colA, count(*) FROM df_flat GROUP BY colA ORDER BY count(*) DESC LIMIT 20").show()查看分组分布，确认是否有极端倾斜的键
4. 观察Spark UI的Stages页面，看哪个Task的处理时间最长、数据量最大，定位瓶颈

内容来源于stack exchange