如何从两个PySpark DataFrame取列合并生成新的DataFrame

你需要的是按行顺序对齐拼接两个DataFrame，PySpark本身没有直接横向拼接（类似pandas的concat(axis=1)）的内置函数，需要通过临时行号关联实现，操作步骤如下：

1. 给两个DataFrame分别添加唯一递增的临时行号列

from pyspark.sql.functions import monotonically_increasing_id

# 给df1添加行号
df1_with_id = df1.withColumn("row_id", monotonically_increasing_id())
# 给df2添加行号
df2_with_id = df2.withColumn("row_id", monotonically_increasing_id())

2. 按临时行号做内连接，拼接完成后删除行号列即可得到目标结果

df_final = df1_with_id.join(df2_with_id, on="row_id", how="inner").drop("row_id")

注意事项：

• 要保证两个DataFrame的行顺序符合拼接预期，如果两个DF本身没有固定排序逻辑，建议先按业务字段排序后再加行号，避免行顺序错乱导致拼接结果不符合预期
• monotonically_increasing_id()生成的是全局唯一且递增的64位整数，只要两个DF行数完全相同，生成的id序列即可一一对应
• 如果你使用的Spark版本>=3.1.0，也可以用row_number()窗口函数生成行号，需要显式指定排序规则，示例如下：

  from pyspark.sql.window import Window
  from pyspark.sql.functions import row_number
  
  # df1按Column_a排序生成行号
  w1 = Window.orderBy("Column_a")
  df1_with_id = df1.withColumn("row_id", row_number().over(w1))
  
  # df2按对应业务字段排序，保证行顺序和df1对齐
  w2 = Window.orderBy("Column_c")
  df2_with_id = df2.withColumn("row_id", row_number().over(w2))
  
  df_final = df1_with_id.join(df2_with_id, on="row_id", how="inner").drop("row_id")
  

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