快速處理Spark中的json文件的方法

我有一組使用嵌套鍵值對的大型壓縮json文件。 json對象中有大約70-80個鍵（和子鍵），但是，我只對幾個鍵感興趣。我想用Spark SQL查詢json文件，只挑出我感興趣的鍵值對，並將它們輸出到一組csv文件。處理一個大小爲170MB的壓縮json文件需要大約5分鐘的時間。我只是想知道是否有任何方法來優化這個過程。或者除了Spark這樣的工作，還有其他更好的工具嗎？謝謝！快速處理Spark中的json文件的方法

下面是我用的是Scala代碼的快照：

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)

來源

2016-08-12 xyin

我;猜大部分時間的推移讀/解壓縮和寫作，這不能並行化。添加分配作業和收集結果的開銷，我的猜測是使用Spark會讓你放慢速度。爲什麼未分析的行的「重新分配」？ –

如果你只是想改變你的數據。你不需要所有的SparkSQL功能。只要堅持RDD的。使用像PlayJson這樣的快速json庫來解析json。修改並轉儲它。 –

除非明確要求，否則請勿對RDD進行重新分區。 –

比方說你的JSON數據看起來像下面，

{ "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", ... }

現在，你可以使用JSON lib和RDD對做轉儲轉儲。

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)

來源

2016-08-12 07:52:47

很簡單的方法來處理JSON：

 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()

見文件：http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets

來源

2016-08-12 12:03:53 pacman

此回答有用嗎？ – pacman

快速處理Spark中的json文件的方法

回答

相關問題