如何在PySpark DataFrame中高效批量复制行以扩充数据集规模

性能问题根因 #

你当前使用的迭代unionAll方案存在两个核心性能问题：

• 每次拼接都会拉长DataFrame的血缘（lineage）链，执行计划会随迭代次数指数级膨胀，调度和解析开销到后期会占据绝大多数运行时间
• 多次拼接的时间复杂度为O(n²)，数据量越大单次迭代的耗时越高，12次迭代后数据量已经到150*2^12=614400行，后续迭代的开销会陡增

高效解决方案 #

根据你使用的框架不同，选择对应方案即可，均可以在几秒到几十秒内完成20M行生成：

分布式场景（PySpark/Spark SQL） #

直接用cross join一次性完成重复，没有多次迭代的血缘膨胀问题，执行计划仅生成一次：

from pyspark.sql.functions import explode, sequence

original_row_count = df.count()
# 向上取整计算需要重复的次数
repeat_times = (20_000_000 + original_row_count - 1) // original_row_count

# 生成对应长度的序列临时表
repeat_df = spark.select(explode(sequence(1, repeat_times)).alias("repeat_id"))
# 跨连接直接得到全量数据，去重辅助列后取刚好20M行
large_df = df.crossJoin(repeat_df) \
             .limit(20_000_000) \
             .drop("repeat_id")

如果后续还要对生成的大表做计算，生成后可以加一句large_df = large_df.checkpoint()切断血缘，避免后续操作重复执行生成逻辑。

单机场景（Pandas） #

不要用循环concat，直接用numpy.tile一次性完成数组复制，再转回DataFrame即可：

import pandas as pd
import numpy as np

original_row_count = len(df)
repeat_times = (20_000_000 + original_row_count - 1) // original_row_count

# 直接复制二维数组的行，速度远快于循环拼接
large_data = np.tile(df.values, (repeat_times, 1))
large_df = pd.DataFrame(large_data, columns=df.columns).head(20_000_000)

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