Apache Sparkの複数行入力でJavaを使用

Question

私はすでにこのサイトで尋ねられている他の類似の質問を見てきましたが、満足のいく回答は得られませんでした。

私はApache sparkとhadoopの初心者です。私の問題は、オンラインショッピングサイトの商品のマルチラインレビューを含む入力ファイル（35GB）があることです。情報は、以下に示すようにファイルに記載されています。

productId: C58500585F 
product: Nun Toy 
product/price: 5.99 
userId: A3NM6WTIAE 
profileName: Heather 
helpfulness: 0/1 
score: 2.0 
time: 1624609 
summary: not very much fun 
text: Bought it for a relative. Was not impressive. 

これは1ブロックのレビューです。そのようなブロックは何千もの空白行で区切られています。私がここから必要としているのは、productId、userId、およびscoreです。したがって、私は必要な行だけをJavaRDDにフィルタリングしました。それは、次のようになります。

productId: C58500585F 
userId: A3NM6WTIAE 
score: 2.0 

コード：

SparkConf conf = new SparkConf().setAppName("org.spark.program").setMaster("local"); 
JavaSparkContext context = new JavaSparkContext(conf); 

JavaRDD<String> input = context.textFile("path"); 

JavaRDD<String> requiredLines = input.filter(new Function<String, Boolean>() { 
public Boolean call(String s) throws Exception { 
if(s.contains("productId") || s.contains("UserId") || s.contains("score") || s.isEmpty()) { 
     return false; 
    } 
    return true; 
} 
}); 

は今、私は方法がわからない1 （キー、値）ペアの一部として、これらの3行を読む必要があるが。 2ブロックのレビューの間には、の空白行しかありません。

私はいくつかのウェブサイトを見てきましたが、私の問題を解決できませんでした。 誰でも私にこれを手伝ってもらえますか？どうもありがとう！詳しい情報が必要な場合はお知らせください。

Answer 1

私の前のコメントから引き続き、textinputformat.record.delimiterをここで使用できます。唯一の区切り文字が空行の場合、値は"\n\n"に設定する必要があります。以下のようなものになります（Scalaでは）

次に

productId: C58500585F 
product: Nun Toy 
product/price: 5.99 
userId: A3NM6WTIAE 
profileName: Heather 
helpfulness: 0/1 
score: 2.0 
time: 1624609 
summary: not very much fun 
text: Bought it for a relative. Was not impressive. 

productId: ABCDEDFG 
product: Teddy Bear 
product/price: 6.50 
userId: A3NM6WTIAE 
profileName: Heather 
helpfulness: 0/1 
score: 2.0 
time: 1624609 
summary: not very much fun 
text: Second comment. 

productId: 12345689 
product: Hot Wheels 
product/price: 12.00 
userId: JJ 
profileName: JJ 
helpfulness: 1/1 
score: 4.0 
time: 1624609 
summary: Summarized 
text: Some text 

コード：

このテストデータを考えてみましょう

import org.apache.hadoop.conf.Configuration 
import org.apache.hadoop.io.{LongWritable, Text} 
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat 
val conf = new Configuration 
conf.set("textinputformat.record.delimiter", "\n\n") 
val raw = sc.newAPIHadoopFile("test.txt", classOf[TextInputFormat], classOf[LongWritable], classOf[Text], conf) 

val data = raw.map(e => { 
    val m = e._2.toString 
    .split("\n") 
    .map(_.split(":", 2)) 
    .filter(_.size == 2) 
    .map(e => (e(0), e(1).trim)) 
    .toMap 

    (m("productId"), m("userId"), m("score").toDouble) 
}) 

出力は次のとおりです。

data.foreach(println) 
(C58500585F,A3NM6WTIAE,2.0) 
(ABCDEDFG,A3NM6WTIAE,2.0) 
(12345689,JJ,4.0) 

正確に確認されませんでしたあなたは出力のために欲しかったので、私はそれを3要素タプルに変えました。また、構文解析ロジックは、必要な場合にはより効率的にすることができますが、これはあなたに何かを働かせる必要があります。