如何在 Spark Scala 中读取带有根元素的多行 json？

Posted 2023-04-15

【中文标题】如何在 Spark Scala 中读取带有根元素的多行 json？

【英文标题】：How to read multiline json with root element in Spark Scala?

【发布时间】：2018-12-12 19:18:02

【问题描述】：

这是一个示例 JSON 文件。我一般都想这样做，比如如果我有根标签，那么如何将 JSON 数据读入 Dataframe 并在控制台中打印。

        "Crimes": [
    
            "ID": 11034701,
            "Case Number": "JA366925",
            "Date": "01/01/2001 11:00:00 AM",
            "Block": "016XX E 86TH PL",
            "IUCR": "1153",
            "Primary Type": "DECEPTIVE PRACTICE",
            "Description": "FINANCIAL IDENTITY THEFT OVER $ 300",
            "Location Description": "RESIDENCE",
            "Arrest": false,
            "Domestic": false,
            "Beat": 412,
            "District": 4,
            "Ward": 8,
            "Community Area": 45,
            "FBI Code": "11",
            "Year": 2001,
            "Updated On": "08/05/2017 03:50:08 PM"
        ,

        
            "ID": 11162428,
            "Case Number": "JA529032",
            "Date": "11/28/2017 09:43:00 PM",
            "Block": "026XX S CALIFORNIA BLVD",
            "IUCR": "5131",
            "Primary Type": "OTHER OFFENSE",
            "Description": "VIOLENT OFFENDER: ANNUAL REGISTRATION",
            "Location Description": "JAIL / LOCK-UP FACILITY",
            "Arrest": true,
            "Domestic": false,
            "Beat": 1034,
            "District": 10,
            "Ward": 12,
            "Community Area": 30,
            "FBI Code": "26",
            "X Coordinate": 1158280,
            "Y Coordinate": 1886310,
            "Year": 2017,
            "Updated On": "02/11/2018 03:54:58 PM",
            "Latitude": 41.843778126,
            "Longitude": -87.694637678,
            "Location": "(41.843778126, -87.694637678)"
        , 
            "ID": 4080525,
            "Case Number": "HL425503",
            "Date": "06/16/2005 09:40:00 PM",
            "Block": "062XX N KIRKWOOD AVE",
            "IUCR": "1365",
            "Primary Type": "CRIMINAL TRESPASS",
            "Description": "TO RESIDENCE",
            "Location Description": "RESIDENCE",
            "Arrest": false,
            "Domestic": false,
            "Beat": 1711,
            "District": 17,
            "Ward": 39,
            "Community Area": 12,
            "FBI Code": "26",
            "X Coordinate": 1145575,
            "Y Coordinate": 1941395,
            "Year": 2005,
            "Updated On": "02/28/2018 03:56:25 PM",
            "Latitude": 41.99518667,
            "Longitude": -87.739863972,
            "Location": "(41.99518667, -87.739863972)"
        , 
            "ID": 4080539,
            "Case Number": "HL422433",
            "Date": "06/15/2005 12:55:00 PM",
            "Block": "042XX S ST LAWRENCE AVE",
            "IUCR": "0460",
            "Primary Type": "BATTERY",
            "Description": "SIMPLE",
            "Location Description": "SCHOOL, PUBLIC BUILDING",
            "Arrest": false,
            "Domestic": false,
            "Beat": 213,
            "District": 2,
            "Ward": 4,
            "Community Area": 38,
            "FBI Code": "08B",
            "X Coordinate": 1180964,
            "Y Coordinate": 1877123,
            "Year": 2005,
            "Updated On": "02/28/2018 03:56:25 PM",
            "Latitude": 41.818075262,
            "Longitude": -87.611675899,
            "Location": "(41.818075262, -87.611675899)"
        
    ]
    

我正在使用此代码。

val conf = new SparkConf().setAppName("demo").setMaster("local"); 
    val sc = new SparkContext(conf);
    val spark = SparkSession.builder().master("local").appName("ValidationFrameWork").getOrCreate()
    val sqlContext = new SQLContext(sc)
    sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
    sc.hadoopConfiguration.set("parquet.enable.summary-metadata", "false")  

    val jsonRDD = sc.wholeTextFiles("D:/FinalScripts/output/Crimes1.json").map(x=>x._2)
    val namesJson = sqlContext.read.json(jsonRDD)
    namesJson.printSchema
     namesJson.registerTempTable("JSONdata")
     val data=sqlContext.sql("select * from JSONdata")
    data.show()

对于此代码，我将一列视为犯罪，并且在一行中完整的数据即将到来。如何忽略根元素并仅获取原始数据。

我什至如何将嵌套的 JSON 读入 Dataframe 并在控制台本身中打印。

【问题讨论】：

你想得到"Crimes"的数组作为输出DataFrame中的行吗？

【参考方案1】：

试试看：

import org.apache.spark.sql.functions._
ds.select(explode($"Crimes") as "exploded").select("exploded.*")

其中 ds 是您从 JSON 记录创建的 Dataset<Row>。

请注意，如果您的数据量很大，Spark 将需要将整个数据保存在内存中，然后再将其展平。

【讨论】：

好的，先生。仅对于此代码，它对我来说工作正常。 :) 非常感谢！