不使用Pandas如何实现两个PySpark DataFrame逐行相减

核心前提 #

PySpark DataFrame 是无固定顺序的分布式数据集，不存在原生的“行对应关系”，要实现按原始位置逐行做差，必须先给每一行生成唯一的位置标识作为关联依据。

推荐实现（适配全量数据场景，无Pandas依赖） #

通过内置的单调递增ID函数给两个表生成行索引，关联后计算差值，全程在Spark分布式计算层完成，不会把数据拉到本地驱动节点，适配大数据量场景：

# 导入SQL函数包
from pyspark.sql import functions as F

# 1. 给两个DataFrame分别添加行ID，同时重命名count列避免关联时列名冲突
df1_marked = df1.withColumn("row_id", F.monotonically_increasing_id()) \
                .withColumnRenamed("count", "count_1")
df2_marked = df2.withColumn("row_id", F.monotonically_increasing_id()) \
                .withColumnRenamed("count", "count_2")

# 2. 按行ID内连接两个表，计算df2与df1的count差值
result_df = df1_marked.join(df2_marked, on="row_id", how="inner") \
                      .withColumn("count", F.col("count_2") - F.col("count_1")) \
                      .select("count")

# 输出验证
result_df.show()

运行后输出结果和预期完全一致：

+-----+
|count|
+-----+
|  200|
|  200|
|  200|
+-----+

小数据量备选方案 #

如果数据量极小（千条级别以内），也可以将两列数据收集到驱动节点直接计算差值再转回DataFrame，写法更简单，但数据量过大会触发驱动节点内存溢出，不推荐生产环境使用：

# 收集两列数据为本地列表
list1 = [row["count"] for row in df1.collect()]
list2 = [row["count"] for row in df2.collect()]
# 逐位计算差值
diff_list = [b - a for a, b in zip(list1, list2)]
# 转回DataFrame
result_df = spark.createDataFrame([(x,) for x in diff_list], schema=["count"])

注意事项 #

• monotonically_increasing_id() 生成的ID保证全局单调递增，单分区内ID按行顺序生成，对于题目给出的示例数据、以及常规从文件/数据源按顺序读取的场景，两个DataFrame同位行的ID完全匹配，不会错位。
• 如果数据经过重分区、shuffle操作后需要严格对齐自定义顺序，建议在生成行号时通过row_number()窗口函数按业务指定的排序字段（比如时间戳、自增主键）排序后生成行ID，避免顺序错位。

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