Spark性能优化：避免collect()处理Databricks海量数据遇瓶颈

一、先优化过滤逻辑，减少源数据量 #

• 优先利用分区过滤：如果id或Topic是表的分区列，先通过分区过滤缩小扫描范围，避免全表扫描。比如读取时直接指定分区，或者先按id过滤再按Topic过滤。
• 广播大列表：如果topics包含大量元素，转为广播变量，减少每个Task的传输开销：

  from pyspark.sql.functions import broadcast
  
  broadcast_topics = spark.sparkContext.broadcast(topics)
  data_filtered = data.filter(
      (data["id"] == event['id']) &
      (data["Topic"].isin(broadcast_topics.value))
  )
  

二、替代collect()，避免一次性拉取所有数据到Driver #

1. 用toLocalIterator()分批处理 #

之前出问题大概率是没在Driver端处理json.loads()，正确做法是拉取单个分区到Driver后再解析JSON：

import json

# 按分区分批拉取，每次仅加载一个分区到Driver内存
for partition in data_filtered.rdd.toLocalIterator():
    for row in partition:
        row_dict = row.asDict()
        # 在Driver端解析目标JSON列
        row_dict["target_json_col"] = json.loads(row_dict["target_json_col"])
        # 执行你的循环处理逻辑

2. 在Executor端批量处理（推荐） #

把数据处理逻辑放在Executor端，避免拉取到Driver，用mapInPandas或foreachBatch实现：

import pandas as pd
import json

def process_batch(batch_df: pd.DataFrame):
    # 批量处理DataFrame，解析JSON并执行业务逻辑
    for _, row in batch_df.iterrows():
        row_dict = row.to_dict()
        row_dict["target_json_col"] = json.loads(row_dict["target_json_col"])
        # 这里直接执行你的处理逻辑，无需返回给Driver
    # 若需输出处理后的数据，返回新DataFrame即可
    return batch_df

# 无返回的处理场景用foreachBatch
data_filtered.write.foreachBatch(lambda df, _: process_batch(df)).mode("append").format("noop").start()

# 需要保留处理后数据的场景用mapInPandas
processed_data = data_filtered.mapInPandas(process_batch, schema=data_filtered.schema)

三、修正Spark配置问题 #

• 配置笔误修正："sark.kryoserializer.buffer.max"改为"spark.kryoserializer.buffer.max"，否则序列化优化不生效。
• 若必须拉取数据到Driver，可适当调大spark.driver.maxResultSize（比如改为8g），但需确保Driver节点有足够内存资源。
• spark.sql.shuffle.partitions调整为Executor总核数的2-3倍（比如100个Executor×4核=400核，设置为800-1200），提升Shuffle效率。

四、先确认过滤后的数据规模 #

执行data_filtered.count()查看实际行数，估算数据大小：

• 如果数据量超过Driver内存上限，必须采用分批或Executor端处理的方案，禁止使用collect()。

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