SPARK-SCALA: обновить дату окончания для идентификатора с новой датой start_date для обновленного соответствующего идентификатора

Question

Я хочу создать новый столбец end_date для id со значением start_date столбца обновленной записи для того же id с использованием Spark Scala

Рассмотрим следующий кадр данных:

+---+-----+----------+
| id|Value|start_date|
+---+---- +----------+
| 1 |   a | 1/1/2018 |
| 2 |   b | 1/1/2018 |
| 3 |   c | 1/1/2018 |
| 4 |   d | 1/1/2018 |
| 1 |   e | 10/1/2018|
+---+-----+----------+

Здесь изначально дата начала id=1 равна 1/1/2018, а value равна a, в то время как 10/1/2018(start_date) значение value из id=1 стало e.поэтому мне нужно заполнить новый столбец end_date и заполнить значение для id=1 в начале значениями 10/1/2018 и NULL для всех других записей для end_date столбца

Результат должен быть таким, как показано ниже:

+---+-----+----------+---------+
| id|Value|start_date|end_date |
+---+---- +----------+---------+
| 1 |   a | 1/1/2018 |10/1/2018|
| 2 |   b | 1/1/2018 |NULL     |
| 3 |   c | 1/1/2018 |NULL     |
| 4 |   d | 1/1/2018 |NULL     |
| 1 |   e | 10/1/2018|NULL     |
+---+-----+----------+---------+

Я использую искру 2.3.

Может кто-нибудь помочь мне здесь, пожалуйста

Answer 1

С оконной функцией «провод»:

val df = List(
  (1, "a", "1/1/2018"),
  (2, "b", "1/1/2018"),
  (3, "c", "1/1/2018"),
  (4, "d", "1/1/2018"),
  (1, "e", "10/1/2018")
).toDF("id", "Value", "start_date")

val idWindow = Window.partitionBy($"id")
  .orderBy($"start_date")

val result = df.withColumn("end_date", lead($"start_date", 1).over(idWindow))
result.show(false)

Выход:

+---+-----+----------+---------+
|id |Value|start_date|end_date |
+---+-----+----------+---------+
|3  |c    |1/1/2018  |null     |
|4  |d    |1/1/2018  |null     |
|1  |a    |1/1/2018  |10/1/2018|
|1  |e    |10/1/2018 |null     |
|2  |b    |1/1/2018  |null     |
+---+-----+----------+---------+