как создать фрейм данных на основе даты первого появления и на основе дополнительных столбцов в каждом столбце идентификатора

Question

я пытаюсь создать фрейм данных со следующим условием: У меня есть несколько идентификаторов, несколько столбцов со значениями по умолчанию (0 или 1) и столбец начальной даты. Я хотел бы получить фрейм данных с появившимися значениями по умолчанию на основе первой начальной даты (default_date) и каждого идентификатора.

оригинальный df выглядит так:

+----+-----+-----+-----+-----------+
|id  |def_a|def_b|deb_c|date       |
+----+-----+-----+-----+-----------+
|  01|    1|    0|    1| 2019-01-31|
|  02|    1|    1|    0| 2018-12-31|
|  03|    1|    1|    1| 2018-10-31|
|  01|    1|    0|    1| 2018-09-30|
|  02|    1|    1|    0| 2018-08-31|
|  03|    1|    1|    0| 2018-07-31|
|  03|    1|    1|    1| 2019-05-31|

вот как я хотел бы иметь:

+----+-----+-----+-----+-----------+
|id  |def_a|def_b|deb_c|date       |
+----+-----+-----+-----+-----------+
|  01|    1|    0|    1| 2018-09-30|
|  02|    1|    1|    0| 2018-08-31|
|  03|    1|    1|    1| 2018-07-31|

я попробовал следующий код:

val w = Window.partitionBy($"id").orderBy($"date".asc) 
val reult = join3.withColumn("rn", row_number.over(w)).where($"def_a" === 1 || $"def_b" === 1 ||$"def_c" === 1).filter($"rn" >= 1).drop("rn")

result.show

Буду признателен за любую помощь

Answer 1

Это должно работать для вас.Сначала назначьте минимальную дату оригинальному df, затем присоедините новый df2 к df.

import org.apache.spark.sql.expressions.Window

val df = Seq(
(1,1,0,1,"2019-01-31"),
(2,1,1,0,"2018-12-31"),
(3,1,1,1,"2018-10-31"),
(1,1,0,1,"2018-09-30"),
(2,1,1,0,"2018-08-31"),
(3,1,1,0,"2018-07-31"),
(3,1,1,1,"2019-05-31"))
.toDF("id"  ,"def_a" , "def_b", "deb_c", "date")

val w = Window.partitionBy($"id").orderBy($"date".asc) 

val df2 = df.withColumn("date", $"date".cast("date"))
            .withColumn("min_date", min($"date").over(w))
            .select("id", "min_date")
            .distinct()

df.join(df2, df("id") === df2("id") && df("date") === df2("min_date"))
.select(df("*"))
.show

И результат должен быть:

+---+-----+-----+-----+----------+
| id|def_a|def_b|deb_c|      date|
+---+-----+-----+-----+----------+
|  1|    1|    0|    1|2018-09-30|
|  2|    1|    1|    0|2018-08-31|
|  3|    1|    1|    0|2018-07-31|
+---+-----+-----+-----+----------+

Кстати, я полагаю, у вас было немногоошибка в ваших ожидаемых результатах.Это (3, 1, 1, 0, 2018-07-31) не (3, 1, 1, 1, 2018-07-31)