scala - 如何在Apache Spark(scala)中迭代RDD

Question

我使用以下命令用一组包含2个字符串[“ filename”，“ content”]的数组填充RDD。

现在，我想遍历所有这些事件，以便对每个文件名和内容进行处理。

val someRDD = sc.wholeTextFiles("hdfs://localhost:8020/user/cloudera/*")

我似乎找不到任何有关如何执行此操作的文档。

所以我想要的是：

foreach occurrence-in-the-rdd{
   //do stuff with the array found on loccation n of the RDD
} 

Answer 1

Spark中的基本操作是map和filter。

val txtRDD = someRDD filter { case(id, content) => id.endsWith(".txt") }

txtRDD现在将仅包含扩展名为“ .txt”的文件

如果您想对这些文件进行字数统计，您可以说

//split the documents into words in one long list
val words = txtRDD flatMap { case (id,text) => text.split("\\s+") }
// give each word a count of 1
val wordT = words map (x => (x,1))  
//sum up the counts for each word
val wordCount = wordsT reduceByKey((a, b) => a + b)

当您需要执行一些昂贵的初始化操作时，您想使用 mapPartitions －例如，如果要使用像Stanford coreNLP工具这样的库进行命名实体识别。

掌握 map， filter， flatMap和 reduce，您就可以熟练掌握Spark。