Spark: правильная схема для загрузки JSON как DataFrame - PullRequest
Новая
       30

Spark: правильная схема для загрузки JSON как DataFrame

0 голосов
/ 12 сентября 2018

У меня есть JSON как 

{ 1234 : "blah1", 9807: "blah2", 467: "blah_k", ...}

, записанный в сжатом файле.Это отображение одного пространства идентификаторов в другое, где ключи int с и значения string с.

Я хочу загрузить его как DataFrame в Spark.

Я загрузил его как 

val df = spark.read.format("json").load("my_id_file.json.gz")

По умолчанию Spark загрузил его со схемой, которая выглядит как 

 |-- 1234: string (nullable = true)
 |-- 9807: string (nullable = true)
 |-- 467: string (nullable = true)

Вместо этого я хочу свой DataFrameчтобы выглядеть как 

+----+------+
|id1 |id2   |
+----+------+
|1234|blah1 |
|9007|blah2 |
|467 |blah_k|    
+----+------+

Итак, я попробовал следующее. 

import org.apache.spark.sql.types._
val idMapSchema = StructType(Array(StructField("id1", IntegerType, true), StructField("id2", StringType, true)))

val df = spark.read.format("json").schema(idMapSchema).load("my_id_file.json.gz")

Однако загруженный фрейм данных выглядит как 

scala> df.show
+----+----+
|id1 |id2 |
+----+----+
|null|null|
+----+----+

Как мне указатьсхема это исправить?Существует ли «чистый» подход к фрейму данных (без создания RDD и последующего создания DataFrame)?

1 Ответ

0 голосов
/ 12 сентября 2018

Один из способов добиться этого - прочитать входной файл как textFile и применить логику анализа в map(), а затем преобразовать результат в кадр данных 

import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer

val rdd = sparkSession.sparkContext.textFile("my_input_file_path")
  .map(row => {
    val list = new ListBuffer[String]()
    val inputJson = new JSONObject(row)

    for (key <- inputJson.keySet()) {
      val resultJson = new JSONObject()
      resultJson.put("col1", key)
      resultJson.put("col2", inputJson.get(key))

      list += resultJson.toString()
    }

    list
  }).flatMap(row => row)

val df = sparkSession.read.json(rdd)
df.printSchema()
df.show(false)

: 

root
 |-- col1: string (nullable = true)
 |-- col2: string (nullable = true)

+----+------+
|col1|col2  |
+----+------+
|1234|blah1 |
|467 |blah_k|
|9807|blah2 |
+----+------+
...