Spark Scala GroupBy 列和总和值

Question

我是 Apache-spark 的新手，最近开始使用 Scala 进行编码。

我有一个包含 4 列的 RDD，如下所示： （第 1 列 - 名称，2 - 标题，3 - 视图，4 - 大小）

aa    File:Sleeping_lion.jpg 1 8030
aa    Main_Page              1 78261
aa    Special:Statistics     1 20493
aa.b  User:5.34.97.97        1 4749
aa.b  User:80.63.79.2        1 4751
af    Blowback               2 16896
af    Bluff                  2 21442
en    Huntingtown,_Maryland  1 0

我想根据 Column Name 进行分组并获得 Column 视图的总和。

应该是这样的：

aa   3
aa.b 2
af   2
en   1

我曾尝试使用groupByKey 和reduceByKey，但我被卡住了，无法继续。

Answer 1

这应该可行，您阅读文本文件，用分隔符分割每一行，使用适当的文件映射到键值并使用 countByKey：

sc.textFile("path to the text file")
 .map(x => x.split(" ",-1))
 .map(x => (x(0),x(3)))
 .countByKey

要完成我的回答，您可以使用 dataframe api 解决问题（如果这对您来说可能取决于 spark 版本），例如：

val result = df.groupBy("column to Group on").agg(count("column to count on"))

另一种可能是使用sql方法：

val df = spark.read.csv("csv path")
df.createOrReplaceTempView("temp_table")
val result = sqlContext.sql("select <col to Group on> , count(col to count on) from temp_table Group by <col to Group on>")

Answer 2

我假设您已经填充了 RDD。

   //For simplicity, I build RDD this way
      val data = Seq(("aa", "File:Sleeping_lion.jpg", 1, 8030),
          ("aa", "Main_Page", 1, 78261),
          ("aa", "Special:Statistics", 1, 20493),
          ("aa.b", "User:5.34.97.97", 1, 4749),
          ("aa.b", "User:80.63.79.2", 1, 4751),
          ("af", "Blowback", 2, 16896),
          ("af", "Bluff", 2, 21442),
          ("en", "Huntingtown,_Maryland", 1, 0))

数据框方法

  val sql = new SQLContext(sc)        
  import sql.implicits._
  import org.apache.spark.sql.functions._

  val df = data.toDF("name", "title", "views", "size")      
  df.groupBy($"name").agg(count($"name") as "") show

**Result** 
+----+-----+
|name|count|    
+----+-----+    
|  aa|    3|    
|  af|    2|   
|aa.b|    2|    
|  en|    1|    
+----+-----+

RDD 方法（CountByKey(...)）

rdd.keyBy(f => f._1).countByKey().foreach(println(_))

RDD 方法（reduceByKey(...)）

rdd.map(f => (f._1, 1)).reduceByKey((accum, curr) => accum + curr).foreach(println(_))

如果其中任何一个都不能解决您的问题，请分享您的确切位置。