如何无需join按列拼接行序一致的PySpark DataFrame？

拥有两个行数相同、主键序列一致的PySpark DataFrame，能否无需join操作（join需逐行匹配主键，开销较高），实现类似pd.concat([df_list], axis=1)的按列拼接？

现有示例DataFrame：
df1:

id  col1    col2    col3
1001    1   0   1
1002    0   1   1
1003    0   0   1

df2:

id  col4    col5
1001    1   0
1002    1   1
1003    0   1

需拼接得到结果：

id  col1    col2    col3    col4    col5
1001    1   0   1   1   0
1002    0   1   1   1   1
1003    0   0   1   0   1

实际场景需处理20个50万行2万列的宽表，最终生成50万行40万列的大表，且无法通过df.toPandas()转Pandas后用pd.concat拼接（内存不足），故寻求低开销的拼接方案。

PySpark没有原生支持类似Pandas的直接按列拼接（分布式环境无法保证行序一致性），但可以通过添加行索引+轻量join的方式实现需求，且开销远低于基于业务主键的join：

1. 为每个DataFrame添加行索引 #

利用row_number()窗口函数生成唯一行序号，因原数据主键序列一致，该序号可保证行的一一对应：

from pyspark.sql import Window
from pyspark.sql.functions import row_number

# 按主键id排序生成窗口（保证行序与原数据一致）
window = Window.orderBy("id")

# 为df1添加临时行索引
df1_with_idx = df1.withColumn("_row_idx", row_number().over(window))
# 为df2添加临时行索引
df2_with_idx = df2.withColumn("_row_idx", row_number().over(window))

2. 基于行索引拼接并清理字段 #

行索引为连续整数，Spark可高效完成匹配，开销远低于业务主键join：

# 按行索引关联，去掉重复的id列和临时索引列
result_df = df1_with_idx.join(df2_with_idx, on="_row_idx", how="inner") \
                        .drop(df2_with_idx.id) \
                        .drop("_row_idx")

3. 多表批量处理优化 #

若需合并20个表，可批量添加索引后用reduce函数依次合并：

from functools import reduce
from pyspark.sql import DataFrame

# 假设所有待合并DataFrame存于df_list中
df_list_with_idx = [df.withColumn("_row_idx", row_number().over(window)) for df in df_list]

# 定义合并逻辑
def merge_dfs(df_a: DataFrame, df_b: DataFrame) -> DataFrame:
    return df_a.join(df_b, on="_row_idx", how="inner").drop(df_b.id)

# 批量合并所有表
final_df = reduce(merge_dfs, df_list_with_idx).drop("_row_idx")

关键说明 #

• 此方案的join开销极低：行索引为连续整数，Spark匹配效率远高于字符串或非连续业务主键；窗口函数生成索引的开销也极小。
• 宽表适配：合并后列数较多（40万列），需提前调整Spark配置，比如增大spark.sql.maxColumns参数值，避免列数超限。

内容的提问来源于stack exchange，提问作者Omer Farooq Ahmed