Разбор JSON в DataFrame Spark в новые столбцы

Question

Фон

У меня есть кадр данных, который выглядит следующим образом:

------------------------------------------------------------------------
|name   |meals                                                         |
------------------------------------------------------------------------
|Tom    |{"breakfast": "banana", "lunch": "sandwich"}                  |
|Alex   |{"breakfast": "yogurt", "lunch": "pizza", "dinner": "pasta"}  |
|Lisa   |{"lunch": "sushi", "dinner": "lasagna", "snack": "apple"}     |
------------------------------------------------------------------------

Получено из следующего:

var rawDf = Seq(("Tom",s"""{"breakfast": "banana", "lunch": "sandwich"}""" ),
  ("Alex", s"""{"breakfast": "yogurt", "lunch": "pizza", "dinner": "pasta"}"""),
  ("Lisa", s"""{"lunch": "sushi", "dinner": "lasagna", "snack": "apple"}""")).toDF("name", "meals")

Я хочу преобразовать его в кадр данныхэто выглядит так:

------------------------------------------------------------------------
|name   |meal       |food                                              |
------------------------------------------------------------------------
|Tom    |breakfast  | banana                                           |
|Tom    |lunch      | sandwich                                         |
|Alex   |breakfast  | yogurt                                           |
|Alex   |lunch      | pizza                                            |
|Alex   |dinner     | pasta                                            |
|Lisa   |lunch      | sushi                                            |
|Lisa   |dinner     | lasagna                                          |
|Lisa   |snack      | apple                                            |
------------------------------------------------------------------------

Я использую Spark 2.1, поэтому я анализирую json с помощью get_json_object. В настоящее время я пытаюсь получить окончательный фрейм данных, используя промежуточный фрейм данных, который выглядит следующим образом:

------------------------------------------------------------------------
|name   |breakfast |lunch    |dinner  |snack                           |
------------------------------------------------------------------------
|Tom    |banana    |sandwich |null    |null                            |
|Alex   |yogurt    |pizza    |pasta   |null                            |
|Lisa   |null      |sushi    |lasagna |apple                           |
------------------------------------------------------------------------

Получено из следующего:

val intermediaryDF = rawDf.select(col("name"),
  get_json_object(col("meals"), "$." + Meals.breakfast).alias(Meals.breakfast),
  get_json_object(col("meals"), "$." + Meals.lunch).alias(Meals.lunch),
  get_json_object(col("meals"), "$." + Meals.dinner).alias(Meals.dinner),
  get_json_object(col("meals"), "$." + Meals.snack).alias(Meals.snack))

Meals определено в другомфайл, который содержит намного больше записей, чем breakfast, lunch, dinner и snack, но выглядит примерно так:

object Meals {
  val breakfast = "breakfast"
  val lunch = "lunch"
  val dinner = "dinner"
  val snack = "snack"
}

Затем я использую intermediaryDF для вычисленияокончательный DataFrame, вот так:

val finalDF = parsedDF.where(col("breakfast").isNotNull).select(col("name"), col("breakfast")).union(
parsedDF.where(col("lunch").isNotNull).select(col("name"), col("lunch"))).union(
parsedDF.where(col("dinner").isNotNull).select(col("name"), col("dinner"))).union(
parsedDF.where(col("snack").isNotNull).select(col("name"), col("snack")))

Моя проблема

Использование промежуточного DataFrame работает, если у меня есть только несколько типов Meals, но у меня фактически есть 40, и перечисление каждого из нихиз них вычислять intermediaryDF нецелесообразно. Мне также не нравится идея вычислять этот DF в первую очередь. Есть ли способ напрямую перейти от моего необработанного кадра данных к окончательному, без промежуточного шага, а также без явного указания регистра для каждого значения в Meals?

Answer 1

Apache Spark предоставляет поддержку для разбора данных json, но для этого необходимо иметь предопределенную схему для разбора ее правильности. Ваши данные JSON являются динамическими, поэтому вы не можете полагаться на схему.

Один из способов сделать это - не разрешать apache spark анализировать данные, но вы можете анализировать их по значению ключа (например, используя что-то вроде Map[String, String], что довольно обобщенно)

Вот что вы можете сделать вместо этого:

Используйте json mapper для scala

// mapper object created on each executor node
  val mapper = new ObjectMapper with ScalaObjectMapper
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
  mapper.registerModule(DefaultScalaModule)

  val valueAsMap = mapper.readValue[Map[String, String]](s"""{"breakfast": "banana", "lunch": "sandwich"}""")

Это даст вам что-то вроде преобразования строки json в Map [String, String]. Это также можно рассматривать как список пары (ключ, значение)

List((breakfast,banana), (lunch,sandwich))

Теперь в игру вступает часть Spark Apache. Определите пользовательскую пользовательскую функцию для анализа строки и вывода списка пар (ключ, значение)

val jsonToArray = udf((json:String) => {
    mapper.readValue[Map[String, String]](json).toList
  })

Примените это преобразование к столбцам «питание» и преобразует его в столбец типа Array. После этого explode для этих столбцов и выберите запись ключа в качестве столбца meal и запись значения в качестве столбца food

val df1 = rowDf.select(col("name"), explode(jsonToArray(col("meals"))).as("meals"))

df1.select(col("name"), col("meals._1").as("meal"), col("meals._2").as("food"))

Отображение последнего кадра данных, который он выводит:

|name|     meal|    food|
+----+---------+--------+
| Tom|breakfast|  banana|
| Tom|    lunch|sandwich|
|Alex|breakfast|  yogurt|
|Alex|    lunch|   pizza|
|Alex|   dinner|   pasta|
|Lisa|    lunch|   sushi|
|Lisa|   dinner| lasagna|
|Lisa|    snack|   apple|
+----+---------+--------+