scala - 比较两个数据帧并更新值

Question

我有两个数据框，如下所示。

val file1 = spark.read.format("csv").option("sep", ",").option("inferSchema", "true").option("header", "true").load("file1.csv")
file1.show()

+---+-------+-----+-----+-------+
| id|   name|mark1|mark2|version|
+---+-------+-----+-----+-------+
|  1| Priya |   80|   99|      0|
|  2| Teju  |   10|    5|      0|
+---+-------+-----+-----+-------+

val file2 = spark.read.format("csv").option("sep", ",").option("inferSchema", "true").option("header", "true").load("file2.csv")
file2.show()

+---+-------+-----+-----+-------+
| id|   name|mark1|mark2|version|
+---+-------+-----+-----+-------+
|  1| Priya |   80|   99|      0|
|  2| Teju  |   70|    5|      0|
+---+-------+-----+-----+-------+

现在我正在比较两个数据帧并过滤掉这样的不匹配值。

val columns = file1.schema.fields.map(_.name)
val selectiveDifferences = columns.map(col => file1.select(col).except(file2.select(col)))
selectiveDifferences.map(diff => {if(diff.count > 0) diff.show})

+-----+
|mark1|
+-----+
|   10|
+-----+

我需要将额外的行添加到数据帧中，1 表示来自数据帧 2 的不匹配值，并像这样更新版本号。

file1.show()

+---+-------+-----+-----+-------+
| id|   name|mark1|mark2|version|
+---+-------+-----+-----+-------+
|  1| Priya |   80|   99|      0|
|  2| Teju  |   10|    5|      0|
|  3| Teju  |   70|    5|      1|
+---+-------+-----+-----+-------+

我正在努力实现上述步骤，这是我的预期输出。任何帮助，将不胜感激。

Answer 1

您可以使用 except 获取最终数据帧和 union如下

val count = file1.count()

import org.apache.spark.sql.expressions._
import org.apache.spark.sql.functions._
file1.union(file2.except(file1)
                  .withColumn("version", lit(1))      //changing the version
                  .withColumn("id", (row_number.over(Window.orderBy("id")))+lit(count))   //changing the id number
  )

lit , row_number和 window函数用于生成 id 和版本

备注 : 使用窗口函数来生成新的 id 使得这个过程效率低下，因为所有的数据都将被收集在一个执行器中来生成新的 id