schnelle Möglichkeit zur Verarbeitung von JSON-Datei in Spark

Question

Ich habe eine Reihe von großen komprimierten JSON-Dateien mit verschachtelten Schlüssel-Wert-Paaren. Es gibt ungefähr 70-80 Schlüssel (und Unterschlüssel) im json-Objekt, aber ich bin nur an wenigen Schlüsseln interessiert. Ich wollte die JSON-Dateien mit Spark SQL abfragen, nur die Schlüssel-Wert-Paare heraussuchen, die mich interessieren, und sie in eine Reihe von CSV-Dateien ausgeben. Es dauert ungefähr 5 Minuten, um eine komprimierte JSON-Datei mit einer Größe von 170 MB zu verarbeiten. Ich frage mich nur, ob es irgendeinen Weg geben könnte, diesen Prozess zu optimieren. Oder gibt es für diesen Job bessere Tools als Spark? Vielen Dank!

Hier ist eine Momentaufnahme des scala Code, den ich verwendet wurde:

val data = sc.textFile("abcdefg.txt.gz") 
// repartition the data 
val distdata = data.repartition(10) 
val dataDF = sqlContext.read.json(distdata) 
// register a temp table 
dataDF.registerTempTable("pixels") 

// query the json file, grab columns of interest 
val query = 
""" 
    |SELECT col1, col2, col3, col4, col5 
    |FROM pixels 
    |WHERE col1 IN (col1_v1, col1_v2, ...) 
""".stripMargin 
val result = sqlContext.sql(query) 

// reformat the timestamps 
val result2 = result.map(
    row => { 
    val timestamp = row.getAs[String](0).stripSuffix("Z").replace("T"," ") 
    Row(timestamp, row(1), row(2), row(3), row(4), row(5), row(6), row(7), 
     row(8), row(9), row(10), row(11)) 
    } 
) 
// output the result to a csv and remove the square bracket in each row 
val output_file = "/root/target" 
result2.map(row => row.mkString(",")).saveAsTextFile(output_file) 

Answer 1

Hier können Sie Ihre JSON-Daten wie folgt: Nun

{ "c1": "timestamp_1", "c2": "12", "c3": "13", "c": "14", "c5": "15", ... } 
{ "c1": "timestamp_1", "c2": "22", "c3": "23", "c": "24", "c5": "25", ... } 
{ "c1": "timestamp_1", "c2": "32", "c3": "33", "c": "34", "c5": "35", ... } 

sieht sagen, Sie eine json lib und RDD die verwenden können, um Führen Sie den Transformationsspeicherauszug aus.

import play.api.libs.json._ 

val data = sc.textFile("abcdefg.txt.gz") 

val jsonData = data.map(line => Json.parse(line)) 

// filter the rdd and just keep the values of interest 
val filteredData = data 
    .filter(json => { 
    val c1 = (json \ "c1").as[String] 
    List[String]("c1_val1", "c2_val2", ...).contains(c1) 
    }) 

    // reformat the timestamps and transform to tuple 
val result2 = filteredData 
    .map(json => { 
    val ts = (json \ "c1").as[String] 
    val tsFormated = ts.stripSuffix("Z").replace("T"," ") 
    (tsFormated, (json \ "c2").as[String], ...) 
    }) 

val output_file = "/root/target" 

result2.saveAsTextFile(output_file) 

Answer 2

Es ist einfach Art und Weise json zu verarbeiten:

 val path = "examples/src/main/resources/people.json" 
     val peopleDF = spark.read.json(path) 

     peopleDF.printSchema() 

     peopleDF.createOrReplaceTempView("people") 

     val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19") teenagerNamesDF.show() 

     val otherPeopleRDD = spark.sparkContext.makeRDD( """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil) val otherPeople = spark.read.json(otherPeopleRDD) otherPeople.show() 

siehe doc: http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets