Spark 2.3：减去数据帧但保留重复值（Scala）答案

【问题标题】：Spark 2.3: subtract dataframes but preserve duplicate values (Scala)Spark 2.3：减去数据帧但保留重复值（Scala）
【发布时间】：2020-01-23 19:49:14
【问题描述】：

从this问题复制示例：作为一个概念示例，如果我有两个数据框：

words     = [the, quick, fox, a, brown, fox]
stopWords = [the, a]

那么我希望输出是任意顺序：

words - stopWords = [quick, brown, fox, fox]

ExceptAll 可以在 2.4 中执行此操作，但我无法升级。链接问题中的答案特定于数据框：

words.join(stopwords, words("id") === stopwords("id"), "left_outer")
     .where(stopwords("id").isNull)
     .select(words("id")).show()

因为您需要知道 pkey 和其他列。

谁能想出一个适用于任何数据框的答案？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

这是为大家准备的一个实现。我已经在 Spark 2.4.2 中测试过，它也应该适用于 2.3（不是 100% 确定）

    val df1 = spark.createDataset(Seq("the","quick","fox","a","brown","fox")).toDF("c1")
    val df2 = spark.createDataset(Seq("the","a")).toDF("c1")

    exceptAllCustom(df1, df2, Seq("c1")).show()


  def exceptAllCustom(df1 : DataFrame, df2 : DataFrame, pks : Seq[String]): DataFrame = {
    val notNullCondition = pks.foldLeft(lit(0==0))((column,cName) => column && df2(cName).isNull)
    val joinCondition = pks.foldLeft(lit(0==0))((column,cName) => column && df2(cName)=== df1(cName))
    val result = df1.join(df2, joinCondition, "left_outer")
       .where(notNullCondition)

    pks.foldLeft(result)((df,cName) => df.drop(df2(cName)))
  }

结果 -

+-----+
|   c1|
+-----+
|quick|
|  fox|
|brown|
|  fox|
+-----+

【讨论】：

当 pkeys 未知时有解决方案吗？该方法应该可以应用于任意 2 个 dfs，例如 2.4 中的 exceptAll
这是可能的，从数据框的架构中读取所有列并使用它。
这对我不起作用。试试df1 = [1,2,2,3,4,5] 和df2 = [1,3,4,5]。结果应该是[2,2]，但这个解决方案返回[ ]。
好的，它在 2.4 中工作，但在 2.3 中你会得到错误：Resolved attribute(s) c1#8 missing from c1#3 in operator !Filter Due to issues.apache.org/jira/browse/SPARK-14948
如果超过 1 列则不起作用。 df1：+---+------+ |id |name | +---+------+ |1 |Beth | |2 |Alan | |3 |Rachel| |1 |Beth | |2 |Alan | +---+------+ df2：+---+------+ |id |name | +---+------+ |3 |Rachel| +---+------+ 预期：+---+----+ |id |name| +---+----+ |2 |Alan| |2 |Alan| |1 |Beth| |1 |Beth| +---+----+ 实际：+---+----+----+ |id |name|id | +---+----+----+ |1 |Beth|null| |2 |Alan|null| |1 |Beth|null| |2 |Alan|null| +---+----+----+

【解决方案2】：

事实证明，使用df1.except(df2) 然后将结果与df1 合并以获取所有重复项更容易。

完整代码：

def exceptAllCustom(df1: DataFrame, df2: DataFrame): DataFrame = {
    val except = df1.except(df2)

    val columns = df1.columns
    val colExpr: Column = df1(columns.head) <=> except(columns.head)
    val joinExpression = columns.tail.foldLeft(colExpr) { (colExpr, p) =>
        colExpr && df1(p) <=> except(p)
    }

    val join = df1.join(except, joinExpression, "inner")

    join.select(df1("*"))
}

【讨论】：