После прочтения многострочного файла JSON в Apache Spark, как получить вложенные атрибуты в виде столбцов?

Question

Я использую Spark 2.3.2 и читаю многострочный файл JSON.Это вывод df.printSchema():

root
 |-- data: struct (nullable = true)
 |    |-- items: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- context: struct (nullable = true)
 |    |    |    |    |-- environment: struct (nullable = true)
 |    |    |    |    |    |-- tag: struct (nullable = true)
 |    |    |    |    |    |    |-- weather: string (nullable = true)
 |    |    |    |    |    |-- weather: struct (nullable = true)
 |    |    |    |    |    |    |-- clouds: double (nullable = true)
 |    |    |    |    |    |    |-- rain: long (nullable = true)
 |    |    |    |    |    |    |-- temp: long (nullable = true)
 |    |    |    |    |-- personal: struct (nullable = true)
 |    |    |    |    |    |-- activity: struct (nullable = true)
 |    |    |    |    |    |    |-- conditions: array (nullable = true)
 |    |    |    |    |    |    |    |-- element: string (containsNull = true)
 |    |    |    |    |    |    |-- kind: string (nullable = true)
 |    |    |    |    |    |-- status: struct (nullable = true)
 |    |    |    |    |    |    |-- speed: double (nullable = true)
 |    |    |    |    |-- timespace: struct (nullable = true)
 |    |    |    |    |    |-- geo: struct (nullable = true)
 |    |    |    |    |    |    |-- coordinates: array (nullable = true)
 |    |    |    |    |    |    |    |-- element: double (containsNull = true)
 |    |    |    |    |    |    |-- type: string (nullable = true)
 |    |    |    |    |    |-- tag: struct (nullable = true)
 |    |    |    |    |    |    |-- season: string (nullable = true)
 |    |    |    |    |    |-- timestamp: string (nullable = true)
 |    |    |    |-- passport: struct (nullable = true)
 |    |    |    |    |-- pid: string (nullable = true)
 |    |    |    |    |-- uid: string (nullable = true)

Видно, что файл JSON имеет вложенную структуру, и не так просто получить определенные вложенные функции, например, сезон, скорость и т. Д.

Вот как я читаю данные:

SparkSession spark = SparkSession.builder()
                                 .config("spark.rdd.compress", "true")
                                 .appName("Test")
                                 .master("local[*]")
                                 .getOrCreate();
df = spark
    .read()
    .option("multiLine", true).option("mode", "PERMISSIVE")
    .json(filePath);

Как мне получить теги timestamp и weather в отдельном наборе данных?

timestamp  weather
...        ...
...        ...

Я пыталсяэто, но это не сработало:

df.registerTempTable("df");
Dataset result = spark.sql("SELECT data.items.element.passport.uid FROM df");

или

Dataset result = df.withColumn("items",
                org.apache.spark.sql.functions.explode(df.col("data.items")))
                .select(df.col("items.context.environment.weather"));

Answer 1

Вы можете прочитать многострочный файл JSON и выбрать вложенные данные, как показано ниже.

//Read multiline json
Dataset<Row> ds = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
        .json("c:\\temp\\test.json");
//print schema
ds.printSchema();
//get weather
Dataset<Row> ds1 = ds.select("data.items.context.environment.weather");
ds1.show(false);
//get timestamp
Dataset<Row> ds2 = ds.select("data.items.context.timestamp");
ds2.show(false);
//get weather and timestamp
Dataset<Row> ds3 = ds.select("data.items.context.environment.weather", "data.items.context.timestamp");
ds3.show(false);

А с Spark 2.4.0 вы можете использовать функции explode и arrays_zip длявзорвать и объединить несколько столбцов

import static org.apache.spark.sql.functions.explode;
import static org.apache.spark.sql.functions.arrays_zip;
import static org.apache.spark.sql.functions.col;

Dataset<Row> ds4 = ds3.withColumn("values", explode(arrays_zip(col("weather"), col("timestamp")))).select(col("values.weather"), col("values.timestamp"));
ds4.show(false);