Вставить два кадра в Scala

Question

У меня есть два источника данных, оба из которых имеют мнения о текущем состоянии одного и того же набора объектов.Любой источник данных может содержать самые последние данные, которые могут быть или не быть с текущей даты.Например:

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 и сортирует по дате и занимает последнюю строку.

Мой вопрос: есть ли идиоматический способ сделать это?

Answer 1

Да, это можно сделать без объединения, используя 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 среди всех дат, подпадающих под одну и ту же id.