Databricks Spark出现[GC (Allocation Failure)]报错求助

• 核心问题：你通过collect()把SKU列表拉到Driver本地循环，每次创建小DataFrame再执行union合并——这种方式会让Driver内存持续累积：每次union都会生成新的逻辑执行计划，Driver需要维护大量中间元数据，加上频繁的小数据处理触发多次垃圾回收（GC），最终导致Driver因内存不足卡住。
• 哪怕总记录数只有2531，781次循环带来的元数据开销、重复计算逻辑，才是引发GC分配失败的关键。

完全抛弃本地循环和collect逻辑，用Spark内置的分布式操作完成分组赋值，代码示例：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

// 假设df_filtered包含SKU、DATEUPDATED、stop及其他业务字段
val new_df = df_filtered
  // 生成唯一分组键：SKU+时间区间的组合
  .withColumn("group_key", concat(col("SKU"), lit("_"), col("DATEUPDATED"), lit("_"), col("stop")))
  // 为每个分组分配唯一counter值
  .withColumn("counter", dense_rank().over(Window.orderBy("group_key")))

如果每个SKU的DATEUPDATED至stop区间是唯一的，也可以用更轻量的分组方式：

val new_df = df_filtered
  .groupBy("SKU", "DATEUPDATED", "stop")
  .agg(collect_list(struct(df_filtered.columns.map(col): _*)).alias("data_rows"))
  .withColumn("counter", monotonically_increasing_id())
  .selectExpr("counter", "explode(data_rows) as row")
  .select("counter", "row.*")

• 移除collect()：避免把分布式数据拉到Driver本地，彻底降低Driver内存负载
• 替换循环+union：用Spark分布式操作一次性完成分组赋值，减少中间元数据的冗余存储
• 利用Spark惰性求值：引擎会自动优化整个执行计划后再批量执行，避免多次小任务触发频繁GC

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