Как читать вложенный JSON в Spark Scala?

Question

Вот мой вложенный JSON-файл.

{
"dc_id": "dc-101",
"source": {
    "sensor-igauge": {
      "id": 10,
      "ip": "68.28.91.22",
      "description": "Sensor attached to the container ceilings",
      "temp":35,
      "c02_level": 1475,
      "geo": {"lat":38.00, "long":97.00}                        
    },
    "sensor-ipad": {
      "id": 13,
      "ip": "67.185.72.1",
      "description": "Sensor ipad attached to carbon cylinders",
      "temp": 34,
      "c02_level": 1370,
      "geo": {"lat":47.41, "long":-122.00}
    },
    "sensor-inest": {
      "id": 8,
      "ip": "208.109.163.218",
      "description": "Sensor attached to the factory ceilings",
      "temp": 40,
      "c02_level": 1346,
      "geo": {"lat":33.61, "long":-111.89}
    },
    "sensor-istick": {
      "id": 5,
      "ip": "204.116.105.67",
      "description": "Sensor embedded in exhaust pipes in the ceilings",
      "temp": 40,
      "c02_level": 1574,
      "geo": {"lat":35.93, "long":-85.46}
    }
  }
}

Как мне прочитать файл JSON в Dataframe с помощью Spark Scala.В файле JSON нет объекта массива, поэтому я не могу использовать explode.Кто-нибудь может помочь?

Answer 1

val df = spark.read.option("multiline", true).json("data/test.json")

df
  .select(col("dc_id"), explode(array("source.*")) as "level1")
  .withColumn("id", col("level1.id"))
  .withColumn("ip", col("level1.ip"))
  .withColumn("temp", col("level1.temp"))
  .withColumn("description", col("level1.description"))
  .withColumn("c02_level", col("level1.c02_level"))
  .withColumn("lat", col("level1.geo.lat"))
  .withColumn("long", col("level1.geo.long"))
  .drop("level1")
  .show(false)

Пример вывода:

+------+---+---------------+----+------------------------------------------------+---------+-----+-------+
|dc_id |id |ip             |temp|description                                     |c02_level|lat  |long   |
+------+---+---------------+----+------------------------------------------------+---------+-----+-------+
|dc-101|10 |68.28.91.22    |35  |Sensor attached to the container ceilings       |1475     |38.0 |97.0   |
|dc-101|8  |208.109.163.218|40  |Sensor attached to the factory ceilings         |1346     |33.61|-111.89|
|dc-101|13 |67.185.72.1    |34  |Sensor ipad attached to carbon cylinders        |1370     |47.41|-122.0 |
|dc-101|5  |204.116.105.67 |40  |Sensor embedded in exhaust pipes in the ceilings|1574     |35.93|-85.46 |
+------+---+---------------+----+------------------------------------------------+---------+-----+-------+

Вместо выбора каждого столбца, вы можете попробовать написать какой-нибудь универсальный UDF для получения всех отдельных столбцов.

Примечание: Протестировано с Spark 2.3

Answer 2

Взять строку в переменную с именем jsonString

import org.apache.spark.sql._
import spark.implicits._
val df = spark.read.json(Seq(jsonString).toDS)
val df1 = df.withColumn("lat" ,explode(array("source.sensor-igauge.geo.lat")))

Вы можете выполнить те же шаги и для других структур - структуры карты / массива