PySpark循环调用union/unionAll报java.lang.StackOverflowError求助

你遇到的StackOverflowError是因为循环调用unionAll会不断累加DataFrame的执行血统（lineage），当迭代次数上万时，Spark的执行计划会变成极深的嵌套树，直接超出JVM的栈容量限制。另外你的代码还有两个严重的性能问题：

• 每次循环都执行dropDuplicates，会触发多次分布式计算，纯粹浪费资源
• 每次循环都写CSV，频繁的IO操作会把整体速度拖得极低

方案1：先收集所有数据再一次性创建DataFrame（数据量不大时） #

如果所有obj的数据总量在Driver内存可承受范围内，直接把所有数据收集到列表，一次性生成DataFrame后再处理：

# 先把所有数据汇总到列表
all_data = []
for page in pages:
    all_data.extend(page['Contents'])

# 一次性创建完整DataFrame
df = spark.createDataFrame(all_data)

# 仅执行一次去重和写入
df.dropDuplicates(col).write.csv(path)

方案2：批量合并，避免血统过长（数据量较大时） #

如果单条obj数据量太大，Driver内存装不下所有数据，就用批量合并的方式，控制每次合并的DataFrame数量，减少执行计划的嵌套深度：

batch_size = 1000  # 可根据集群内存情况调整大小
dfs = []

for page in pages:
    for obj in page['Contents']:
        df = spark.createDataFrame(obj)
        dfs.append(df)
        # 达到批量阈值就合并一次
        if len(dfs) >= batch_size:
            temp_df = dfs[0]
            for sub_df in dfs[1:]:
                temp_df = temp_df.unionAll(sub_df)
            dfs = [temp_df]

# 合并剩余的DataFrame
if dfs:
    final_df = dfs[0]
    for sub_df in dfs[1:]:
        final_df = final_df.unionAll(sub_df)
    final_df.dropDuplicates(col).write.csv(path)

方案3：用迭代器直接创建DataFrame（最优大数据方案） #

Spark的createDataFrame支持传入迭代器，不需要提前把所有数据加载到Driver内存，这是处理超大分页数据的最优方式：

# 定义迭代器，按需生成数据
def data_generator():
    for page in pages:
        yield from page['Contents']

# 基于迭代器创建DataFrame
df = spark.createDataFrame(data_generator())

# 一次完成去重和写入
df.dropDuplicates(col).write.csv(path)

• 杜绝循环union：尽量减少DataFrame的union次数，避免执行计划的血统嵌套过深
• 延迟重计算：只在所有数据合并完成后执行一次去重和写入，避免重复计算和IO
• 平衡内存与复杂度：用迭代器或批量处理，既不压垮Driver内存，也能控制执行计划的深度

内容的提问来源于stack exchange，提问作者Hari Prasad Eluri