非常に大きなJSONファイルをSparkで変換する最速の方法は何ですか？

Question

1つのJSONオブジェクトがたくさんあるかなり大きなJSONファイル（Amazon product data）があります。これらのJSONオブジェクトには、特定のトレーニングタスクのために前処理したいテキストが含まれていますが、ここで高速化する必要があるのは前処理です。一つのJSONオブジェクトは、次のようになります。

{ 
    "reviewerID": "A2SUAM1J3GNN3B", 
    "asin": "0000013714", 
    "reviewerName": "J. McDonald", 
    "helpful": [2, 3], 
    "reviewText": "I bought this for my husband who plays the piano. He is having a wonderful time playing these old hymns. The music is at times hard to read because we think the book was published for singing from more than playing from. Great purchase though!", 
    "overall": 5.0, 
    "summary": "Heavenly Highway Hymns", 
    "unixReviewTime": 1252800000, 
    "reviewTime": "09 13, 2009" 
} 

タスクは、各JSONオブジェクトからreviewTextを抽出し、などlemmatizingのようないくつかの前処理を行うことであろう

私の問題は、私は、私が使用することができます方法がわからないということですこのタスクをクラスタ上で高速化するためにSparkを使用してください。JSONファイルをオブジェクトごとのストリームとして読み込み、メインタスクを並列化できるかどうかは実際にはわかりません。

これを開始するにはどうすればよいでしょうか？

Answer 1

1行に1つのJSONオブジェクトがあるので、RDDのtextFileを使用して、RDD[String]の行を取得できます。次にmapを使用して、json4sのようなものを使用してJSONオブジェクトを解析し、必要なフィールドを抽出します。あなたはJSONデータセットを使用して、reviewText列の値を取得するために、単純なSQLクエリを実行することができます

import org.json4s._ 
import org.json4s.jackson.JsonMethods._ 
implicit def formats = DefaultFormats 

val r = sc.textFile("input_path").map(l => (parse(l) \ "reviewText").extract[String]) 

Answer 2

私はDataframeにJSONデータをロードし、必要なフィールドを選択します。マップを使用して、lemmatisingのような前処理を適用することもできます。

Answer 3

：

コード全体の意志は、このような単純なに見えます

（SparkContextscとしてあなたが持っていると仮定）

SparkSQLドキュメント（ http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets）での実施例から内蔵

// A JSON dataset is pointed to by path. 
// The path can be either a single text file or a directory storing text files. 
val path = "path/reviews.json" 
val people = sqlContext.read.json(path) 
// Register this DataFrame as a table. 
people.registerTempTable("reviews") 
val reviewTexts = sqlContext.sql("SELECT reviewText FROM reviews")