Spark周级聚合中窗口函数引发的分区及性能问题咨询

原有每日作业逻辑：按user和day维度聚合数据，写入S3的year=<>/month=<>/day=<>分区，单日常规处理时长约30分钟。
现需调整为周级处理逻辑：一次性处理7天数据，按user和week维度聚合，最终写入year=<>/month=<>/week=<>或year_month=<>/week=<>目录。
数据规模：每日输入数据600GB，周级输入Parquet数据达3.5TB。
当前问题：使用以下代码时，单日数据处理时长超7小时，7天数据处理直接因“Stage failed, executor heartbeat timed out”失败：

# year_month存储最新年月，覆盖年末/月末跨年月的同周数据
df = df.withColumn('year_month', expr("max(concat(year(timestamp_field), lpad(month(timestamp_field), 2, '0'))) over()")) \
      .withColumn('week', weekofyear("timestamp_field"))

核心疑问：

1. 处理7天（或N天）数据时，上述无分区窗口函数是否会将全量数据拉至单个分区引发OOM？
2. 该逻辑能否并行处理？
3. 有哪些优化方案？

关于无分区窗口函数的问题 #

• 是的，你使用的over()未指定PARTITION BY，属于全局窗口函数，Spark会把全量数据shuffle到单个分区中计算全局最大值。3.5TB的数据集中到一个分区，必然触发OOM，且单分区计算完全无法利用集群并行能力，这就是单日处理时长暴增、周级任务直接失败的核心原因。
• 这种全局窗口逻辑本身无法并行处理，因为要计算全局最大值，必须将所有数据汇总到一个节点才能完成，天然串行。

优化方案 #

1. 重构year_month计算逻辑，规避全局窗口 #

你的需求是用同周内的最新年月作为year_month，可换一种并行化方式实现：

# 先给每条数据打上当前年月标签
df = df.withColumn('current_year_month', expr("concat(year(timestamp_field), lpad(month(timestamp_field), 2, '0'))"))
# 按week分区，计算每个周的最大年月，再关联回原表
week_max_ym = df.groupBy('week').agg(max('current_year_month').alias('year_month'))
df = df.join(week_max_ym, on='week', how='left')

这种方式下，数据会按week分区并行计算，每个分区仅处理对应周的数据，完全利用集群并行能力，不会出现单分区过载的情况。

2. 作业层面通用优化 #

• 调整Spark资源配置：针对3.5TB数据，适当调大executor内存（如--executor-memory 32G）、executor核数（如--executor-cores 8），同时增加executor数量；另外调大spark.sql.shuffle.partitions（建议设为2000-4000，根据数据量调整），避免shuffle后分区过大。
• 输入数据筛选：如果S3上的每日数据按year/month/day分区存储，周级处理时仅读取对应7天的分区，避免扫描全量数据；Parquet默认开启的谓词下推也能进一步减少读取数据量。
• 写入优化：写入S3时开启spark.sql.parquet.compression.codec=snappy压缩，减少数据传输和存储量；用coalesce或repartition调整输出分区数，避免生成过多小文件影响后续读取性能。

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