如何从指定JSON提取data字段生成Spark DataFrame（Scala）

我需要从如下JSON结构中提取data字段内容，转换为带指定列名的Spark DataFrame：

{
  "meta" : {},
  "data" : [ 
    [ 1, "a1", "b1" ],
    [ 2, "a2", "b2" ],
    [ 3, "a3", "b3" ]
  ]
}

我用Scala开发，当前写了如下函数：

def GetUrlContentJson(url: String): DataFrame ={
    val result = scala.io.Source.fromURL(url).mkString
    val jsonResponseOneLine = result.toString().stripLineEnd    
    val jsonRdd = spark.sparkContext.parallelize(jsonResponseOneLine :: Nil)
    val jsonDf = spark.read.json(spark.createDataset(jsonRdd))
    return jsonDf
}  

但得到的结果是：

+-------+-------+
|   data|   meta|
+-------+-------+
| [[1...|     {}|
+-------+-------+

期望得到带指定列名的结构化结果：

+----+------+------+
| id |   A  |   B  |
+----+------+------+
|  1 |   a1 |  b1  |
+----+------+------+
|  2 |   a2 |  b2  |
+----+------+------+
|  3 |   a3 |  b3  |
+----+------+------+

方法一：使用Spark DataFrame API #

先解析出data数组，将数组中的子数组展开为单独行，再拆分元素映射为指定列：

import org.apache.spark.sql.functions.{explode, col}

def GetUrlContentJson(url: String): DataFrame ={
    // 读取URL中的JSON内容并转为Dataset
    val result = scala.io.Source.fromURL(url).mkString
    val jsonDs = spark.createDataset(Seq(result))
    val jsonDf = spark.read.json(jsonDs)
    
    // 提取data数组并展开，拆分元素为指定列
    jsonDf.select(explode(col("data")).alias("data_arr"))
          .select(
              col("data_arr")(0).alias("id"),
              col("data_arr")(1).alias("A"),
              col("data_arr")(2).alias("B")
          )
}

方法二：使用Spark SQL #

通过创建临时视图，用SQL语句完成提取和列映射：

def GetUrlContentJson(url: String): DataFrame ={
    val result = scala.io.Source.fromURL(url).mkString
    val jsonDs = spark.createDataset(Seq(result))
    val jsonDf = spark.read.json(jsonDs)
    
    // 创建临时视图
    jsonDf.createOrReplaceTempView("json_data")
    
    // 执行SQL转换结构
    spark.sql("""
        SELECT 
            data_arr[0] AS id,
            data_arr[1] AS A,
            data_arr[2] AS B
        FROM json_data
        LATERAL VIEW explode(data) exploded_table AS data_arr
    """)
}

补充说明 #

• 必须导入explode等必要函数，避免编译错误
• 直接将JSON字符串转为Dataset比RDD写法更简洁高效
• 若data子数组长度固定，直接通过索引取值即可；长度不固定时可根据实际逻辑调整

内容的提问来源于stack exchange，提问作者Don Dev