Заполнение пропущенных значений в строках с помощью Apache spark

Question

У меня есть особое требование, чтобы заполнить все значения (категории) в столбце.Например, как показано в таблице ниже.Я хочу заполнить категории «НЕИЗВЕСТНО» и «НАЗНАЧЕНО» для кода HL_14108.

val df = Seq(
("HL_13203","DELIVERED",3226), 
("HL_13203","UNSEEN",249),     
("HL_13203","UNDELIVERED",210),
("HL_13203","ASSIGNED",2),    
("HL_14108","DELIVERED",3083), 
("HL_14108","UNDELIVERED",164),
("HL_14108","PICKED",1)).toDF("code","status","count")

Ввод:

+--------+-----------+-----+
|    code|     status|count|
+--------+-----------+-----+
|HL_13203|  DELIVERED| 3226|
|HL_13203|     UNSEEN|  249|
|HL_13203|UNDELIVERED|  210|
|HL_13203|   ASSIGNED|    2|
|HL_14108|  DELIVERED| 3083|
|HL_14108|UNDELIVERED|  164|
|HL_14108|     PICKED|    1|
+--------+-----------+-----+

Ожидаемый вывод:

+--------+-----------+-----+
|    code|     status|count|
+--------+-----------+-----+
|HL_13203|  DELIVERED| 3226|
|HL_13203|     UNSEEN|  249|
|HL_13203|UNDELIVERED|  210|
|HL_13203|   ASSIGNED|    2|
|HL_13203|     PICKED|    0|
|HL_14108|  DELIVERED| 3083|
|HL_14108|UNDELIVERED|  164|
|HL_14108|     PICKED|    1|
|HL_14108|     UNSEEN|    0|
|HL_14108|   ASSIGNED|    0|
+--------+-----------+-----+

Я хочу добавить недостающие строки категорий для каждого кода.Какой будет правильный подход, чтобы сделать это в Apache Spark?

Answer 1

Сначала создайте новый фрейм данных со всеми возможными комбинациями столбцов code и status.Это можно сделать по-разному, но наиболее простым является перекрестное соединение:

val states = df.select("status").dropDuplicates()
val codes = df.select("code").dropDuplicates()
val df2 = codes.crossJoin(states)

A лучший подход - сначала идентифицировать все возможные состояния, а затем использовать explode иtypedLit (доступно из Spark версии 2.2+).Это приведет к тому же кадру данных:

val states = df.select("status").dropDuplicates().as[String].collect()
val codes = df.select("code").dropDuplicates()
val df2 = codes.withColumn("status", explode(typedLit(states)))

Для более старых версий Spark ту же функциональность, что и typedLit, можно получить с помощью array(states.map(lit(_)): _*).

---

Затем, join этот новый фрейм данных со старым для получения столбца count.Строки без значения count будут NaN, поэтому na.fill(0) используется для установки их на 0:

df2.join(df, Seq("code", "status"), "left").na.fill(0)

Результирующий кадр данных:

+--------+-----------+-----+
|    code|     status|count|
+--------+-----------+-----+
|HL_13203|UNDELIVERED|  210|
|HL_13203|   ASSIGNED|    2|
|HL_13203|     UNSEEN|  249|
|HL_13203|     PICKED|    0|
|HL_13203|  DELIVERED| 3226|
|HL_14108|UNDELIVERED|  164|
|HL_14108|   ASSIGNED|    0|
|HL_14108|     UNSEEN|    0|
|HL_14108|     PICKED|    1|
|HL_14108|  DELIVERED| 3083|
+--------+-----------+-----+