如何计算不同数据框的列之间的数值差异？答案

【问题标题】：How to compute the numerical difference between columns of different dataframes?如何计算不同数据框的列之间的数值差异？
【发布时间】：2019-10-23 14:56:49
【问题描述】：

给定两个具有相同列数和行数的 spark 数据帧 A 和 B，我想计算两个数据帧之间的数值差异并将其存储到另一个数据帧（或另一个数据结构可选）中。

例如让我们有以下数据集

数据帧 A：

+----+---+
|  A | B |
+----+---+
|   1|  0|
|   1|  0|
+----+---+

数据帧 B：

----+---+
|  A | B |
+----+---+
|   1| 0 |
|   0| 0 |
+----+---+

如何获得B-A，即


+----+---+
| c1 | c2|
+----+---+
|   0| 0 |
|  -1| 0 |
+----+---+

在实践中，实际数据帧具有相应的行数和 50 多列，需要计算差异。 Spark/Scala 的实现方式是什么？

【问题讨论】：

n列的一般情况还没有解决，你有解决办法吗？

标签： apache-spark apache-spark-sql

【解决方案1】：

我能够使用以下方法解决此问题。此代码可以处理任意数量的列。您只需相应地更改输入 DF。

import org.apache.spark.sql.Row

val df0 = Seq((1, 5), (1, 4)).toDF("a", "b")
val df1 = Seq((1, 0), (3, 2)).toDF("a", "b")

val columns = df0.columns
    val rdd = df0.rdd.zip(df1.rdd).map {
      x =>
        val arr = columns.map(column =>
          x._2.getAs[Int](column) - x._1.getAs[Int](column))
        Row(arr: _*)
    }

spark.createDataFrame(rdd, df0.schema).show(false)

生成的输出：

df0=>
+---+---+
|a  |b  |
+---+---+
|1  |5  |
|1  |4  |
+---+---+
df1=>
+---+---+
|a  |b  |
+---+---+
|1  |0  |
|3  |2  |
+---+---+
Output=>
+---+---+
|a  |b  |
+---+---+
|0  |-5 |
|2  |-2 |
+---+---+

【讨论】：

这个答案很好，但不要概括为 n 列的情况。如果你管理你概括它我会接受答案。
我已经更新了代码。现在是通用的。请检查并让我知道它是否有效。另外，请接受答案。

【解决方案2】：

如果您的 df A 与 df B 相同，您可以尝试以下方法。我不知道这是否适用于大型数据集，最好有 id 来加入而不是使用 monotonically_increasing_id() 创建它。

  import spark.implicits._
  import org.apache.spark.sql.functions._

  val df0 = Seq((1, 0), (1, 0)).toDF("a", "b")
  val df1 = Seq((1, 0), (0, 0)).toDF("a", "b")

  // new cols names
  val colNamesA = df0.columns.map("A_" + _)
  val colNamesB = df0.columns.map("B_" + _)

  // rename cols and add id
  val dfA = df0.toDF(colNamesA: _*)
    .withColumn("id", monotonically_increasing_id())
  val dfB = df1.toDF(colNamesB: _*)
    .withColumn("id", monotonically_increasing_id())

  dfA.show()
  dfB.show()

  // get columns without id
  val dfACols = dfA.columns.dropRight(1).map(dfA(_))
  val dfBCols = dfB.columns.dropRight(1).map(dfB(_))

  // diff between cols
  val calcCols = (dfACols zip dfBCols).map(s=>s._2-s._1)

  // join dfs
  val joined = dfA.join(dfB, "id")
  joined.show()

  calcCols.foreach(_.explain(true))

  joined.select(calcCols:_*).show()

    +---+---+---+
    |A_a|A_b| id|
    +---+---+---+
    |  1|  0|  0|
    |  1|  0|  1|
    +---+---+---+

    +---+---+---+
    |B_a|B_b| id|
    +---+---+---+
    |  1|  0|  0|
    |  0|  0|  1|
    +---+---+---+

    +---+---+---+---+---+
    | id|A_a|A_b|B_a|B_b|
    +---+---+---+---+---+
    |  0|  1|  0|  1|  0|
    |  1|  1|  0|  0|  0|
    +---+---+---+---+---+

    (B_a#26 - A_a#18)
    (B_b#27 - A_b#19)

    +-----------+-----------+
    |(B_a - A_a)|(B_b - A_b)|
    +-----------+-----------+
    |          0|          0|
    |         -1|          0|
    +-----------+-----------+

【讨论】：