代码之家 › 专栏 › 技术社区 › Jeremy

在Scala中向上插入两个数据帧

apache-spark scala

1

Jeremy · 技术社区 · 7 年前

我有两个数据源,它们都对同一组实体的当前状态有看法。任何一个数据源都可能包含最新的数据,这些数据可能来自当前日期,也可能不来自当前日期。例如:

val df1 = Seq((1, "green", "there", "2018-01-19"), (2, "yellow", "there", "2018-01-18"), (4, "yellow", "here", "2018-01-20")).toDF("id", "status", "location", "date")

val df2 = Seq((2, "red", "here", "2018-01-20"), (3, "green", "there", "2018-01-20"), (4, "green", "here", "2018-01-19")).toDF("id", "status", "location", "date")

df1.show
+---+------+--------+----------+
| id|status|location|      date|
+---+------+--------+----------+
|  1| green|   there|2018-01-19|
|  2|yellow|   there|2018-01-18|
|  4|yellow|    here|2018-01-20|
+---+------+--------+----------+

df2.show
+---+------+--------+----------+
| id|status|location|      date|
+---+------+--------+----------+
|  2|   red|    here|2018-01-20|
|  3| green|   there|2018-01-20|
|  4| green|    here|2018-01-19|
+---+------+--------+----------+

+---+------+--------+----------+
| id|status|location|      date|
+---+------+--------+----------+
|  1| green|   there|2018-01-19|
|  2|   red|    here|2018-01-20|
|  3| green|   there|2018-01-20|
|  4|yellow|    here|2018-01-20|
+---+------+--------+----------+

我的方法似乎很有效,将两个表连接起来,然后根据日期执行一种自定义合并操作:

val joined = df1.join(df2, df1("id") === df2("id"), "outer")
+----+------+--------+----------+----+------+--------+----------+
|  id|status|location|      date|  id|status|location|      date|
+----+------+--------+----------+----+------+--------+----------+
|   1| green|   there|2018-01-19|null|  null|    null|      null| 
|null|  null|    null|      null|   3| green|   there|2018-01-20| 
|   4|yellow|    here|2018-01-20|   4|yellow|    here|2018-01-20|
|   2|yellow|   there|2018-01-18|   2|   red|    here|2018-01-20|
+----+------+--------+----------+----+------+--------+----------+

val weirdCoal(name: String) = when(df1("date") > df2("date") || df2("date").isNull, df1(name)).otherwise(df2(name)) as name

val ouput = joined.select(df1.columns.map(weirdCoal):_*)
+---+------+--------+----------+
| id|status|location|      date|
+---+------+--------+----------+
|  1| green|   there|2018-01-19|
|  2|   red|    here|2018-01-20|
|  3| green|   there|2018-01-20|
|  4|yellow|    here|2018-01-20|
+---+------+--------+----------+

这就是我期望的结果。

我还可以看到通过某种联合/聚合方法或通过一个按id划分、按日期排序并取最后一行的窗口来实现这一点。

我的问题是:有没有一种惯用的方法?

1 回复 | 直到 7 年前

1

vdep 7 年前

是的,不用连接就可以完成 Window 功能:

df1.union(df2)
  .withColumn("rank", rank().over(Window.partitionBy($"id").orderBy($"date".desc)))
  .filter($"rank" === 1)
  .drop($"rank")
  .orderBy($"id")
  .show

+---+------+--------+----------+
| id|status|location|      date|
+---+------+--------+----------+
|  1| green|   there|2018-01-19|
|  2|   red|    here|2018-01-20|
|  3| green|   there|2018-01-20|
|  4|yellow|    here|2018-01-20|
+---+------+--------+----------+

上面的代码按照 id date 身份证件 .