Spark：将 JSON 加载为 DataFrame 的正确模式答案

【问题标题】：Spark: correct schema to load JSON as DataFrameSpark：将 JSON 加载为 DataFrame 的正确模式
【发布时间】：2019-02-16 12:04:23
【问题描述】：

我有一个类似的 JSON

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

写入压缩文件。它是一个 ID 空间到另一个 ID 空间的映射，其中键是 ints，值是 strings。

我想在 Spark 中将其加载为 DataFrame。

我把它加载为，

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）？

【问题讨论】：

标签： json apache-spark apache-spark-sql schema

【解决方案1】：

实现此目的的一种方法是将输入文件读取为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 |
+----+------+

【讨论】：