比较两个数据框并更新值
Posted
技术标签:
【中文标题】比较两个数据框并更新值【英文标题】:Compare two dataframes and update the values 【发布时间】:2018-04-20 09:09:57 【问题描述】:我有两个如下的数据框。
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|
+---+-------+-----+-----+-------+
我正在努力实现上述步骤,这是我的预期输出。任何帮助将不胜感激。
【问题讨论】:
【参考方案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
【讨论】:
以上是关于比较两个数据框并更新值的主要内容,如果未能解决你的问题,请参考以下文章