如何在 Spark 2.3.0 中减去两个保持重复的 DataFrame

【问题标题】：How to subtract two DataFrames keeping duplicates in Spark 2.3.0如何在 Spark 2.3.0 中减去两个保持重复的 DataFrame
【发布时间】：2019-04-26 18:31:32
【问题描述】：

Spark 2.4.0 引入了新的便捷函数exceptAll，它允许减去两个数据帧，保持重复。

例子

  val df1 = Seq(
    ("a", 1L),
    ("a", 1L),
    ("a", 1L),
    ("b", 2L)
  ).toDF("id", "value")
  val df2 = Seq(
    ("a", 1L),
    ("b", 2L)
  ).toDF("id", "value")

df1.exceptAll(df2).collect()
// will return

Seq(("a", 1L),("a", 1L))

但是我只能使用 Spark 2.3.0。

仅使用 Spark 2.3.0 中的函数来实现此功能的最佳方法是什么？

【问题讨论】：

标签： apache-spark apache-spark-sql

【解决方案1】：

一种选择是使用row_number 生成一个序列号列并在left join 上使用它来获取丢失的行。

此处显示的 PySpark 解决方案。

 from pyspark.sql.functions import row_number
 from pyspark.sql import Window
 w1 = Window.partitionBy(df1.id).orderBy(df1.value)
 w2 = Window.partitionBy(df2.id).orderBy(df2.value)
 df1 = df1.withColumn("rnum", row_number().over(w1))
 df2 = df2.withColumn("rnum", row_number().over(w2))
 res_like_exceptAll = df1.join(df2, (df1.id==df2.id) & (df1.val == df2.val) & (df1.rnum == df2.rnum), 'left') \
                         .filter(df2.id.isNull()) \ #Identifies missing rows 
                         .select(df1.id,df1.value)
 res_like_exceptAll.show()

【讨论】：