字数（频率） spark rdd scala答案

【问题标题】：word count(frequency) spark rdd scala字数（频率） spark rdd scala
【发布时间】：2016-04-28 14:46:27
【问题描述】：

如果我有一个跨集群的 rdd 并且我想做字数统计

不仅计算出现次数，我想得到频率，定义为计数/总计数

在 scala 中最好和最有效的方法是什么？如何在一个工作流程中同时进行缩减工作和计算总数？

顺便说一句，我知道纯字数可以通过这种方式完成。

text_file = spark.textFile("hdfs://...")
counts = text_file.flatMap(lambda line: line.split(" ")) \
             .map(lambda word: (word, 1)) \
             .reduceByKey(lambda a, b: a + b)
counts.saveAsTextFile("hdfs://...")

但是如果我使用聚合有什么区别呢？就 Spark 工作流程而言

    val result = pairs
       .aggregate(Map[String, Int]())((acc, pair) => 
            if(acc.contains(pair._1)) 
                acc ++ Map[String, Int]((pair._1, acc(pair._1)+1)) 
            else 
                acc ++ Map[String, Int]((pair._1, pair._2))
          , 
          (a, b) => 
              (a.toSeq ++ b.toSeq)
                 .groupBy(_._1)
                 .mapValues(_.map(_._2).reduce(_ + _))
       )

【问题讨论】：

标签： scala apache-spark rdd word-count

【解决方案1】：

你可以用这个

val total = counts.map(x => x._2).sum() 
val freq  = counts.map(x => (x._1, x._2/total))

还有 Accumulator 的概念，它是一个只写变量，您可以使用它来避免使用 sum() 操作，但您的代码需要进行大量更改。

【讨论】：