使用 java 在 Apache Spark 中进行多行输入答案

【问题标题】：Multi-line input in Apache Spark using java使用 java 在 Apache Spark 中进行多行输入
【发布时间】：2016-10-14 07:55:44
【问题描述】：

我查看了该网站上已经提出的其他类似问题，但没有得到满意的答案。

我是 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.

这是一个审查块。有数千个这样的块由空行分隔。我从这里需要的是 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;
}
});

现在，我需要将这三行作为 (key, value) 对的一部分来阅读，我不知道该怎么做。两块评论之间只会有一条空白线。

我查看了几个网站，但没有找到解决问题的方法。有人可以帮我吗？非常感谢！如果您需要更多信息，请告诉我。

【问题讨论】：

你有没有想过和textinputformat.record.delimiter一起玩？像this 这样的东西。这样做可以让您获得一个 RDD，其中每条记录都由整个文本块组成。
@Student : 块字段(productid,product...etc) 是否被任何分隔符分割？
@Student：你还需要map而不是filter
@Shankar 不，它们唯一的分隔符是它们位于不同的行上。因此它们仅由新行分隔符分隔，没有其他特殊分隔符。
@Junjun Olympia 我已经研究过了，但正如我在问题中所说，没有特殊的分隔符。块仅由空行分隔。

标签： hadoop apache-spark mapreduce multiline

【解决方案1】：

从我之前的 cmets 继续，textinputformat.record.delimiter 可以在这里使用。如果唯一的分隔符是空行，则值应设置为"\n\n"。

考虑这个测试数据：

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

然后代码（在 Scala 中）看起来像：

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 元素元组。此外，如果您需要，解析逻辑肯定可以提高效率，但这应该会给您一些工作。

【讨论】：