PySpark技术问询：基于关联DataFrame的字段匹配为目标DataFrame新增unique_ID列

针对你需要处理数百万行、20列数据的场景，我整理了一套兼顾性能和需求的实现方案，完全贴合你的预期结果：

完整代码实现 #

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 1. 为df_b添加临时序号，保留原始行顺序（Spark分布式环境下默认不保证行序）
df_b_with_order = df_b.withColumn("temp_order", F.monotonically_increasing_id())

# 2. 广播df_a（如果df_a是维度表、数据量远小于df_b，这一步能大幅减少shuffle，提升性能）
df_a_broadcast = F.broadcast(df_a)

# 3. 通过val_1和val_2组合关联，获取对应的unique_ID
joined_df = df_b_with_order.join(df_a_broadcast, on=["val_1", "val_2"], how="left")

# 4. 生成连续的sl.no序号列
final_df = joined_df.withColumn(
    "sl.no",
    F.row_number().over(Window.orderBy("temp_order"))
).drop("temp_order")

# 查看结果
final_df.show()

关键步骤说明 #

• 保留原始行顺序：Spark是分布式计算框架，默认不会保留DataFrame的行顺序，所以我们用monotonically_increasing_id()生成一个全局唯一的递增临时ID，确保后续生成的sl.no完全匹配df_b的原始行顺序。
• 广播优化：针对数百万行的df_b，广播较小的df_a（通常这种映射表都是维度表，数据量不大）可以让每个Executor都持有df_a的副本，避免大规模数据shuffle，这是处理大数据量时的核心性能优化点。
• 连续序号生成：通过窗口函数row_number()按照临时ID排序，生成连续的sl.no，完全符合你预期的序号格式。

这个方案也能完美适配你20列的数据集，join操作只会用到val_1和val_2两个关联字段，其他列会自动保留在最终结果中。

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