Scala spark 按键归约并找到共同值(value)

Question

我有一个 csv 数据文件，作为 sequenceFile 存储在 HDFS 上，格式为 name, zip, country, fav_food1, fav_food2, fav_food3, fav_colour 。可能有许多同名的条目，我需要找出他们最喜欢的食物是什么(即计算所有记录中具有该名称的所有食物条目并返回最受欢迎的条目。我是 Scala 和 Spark 的新手并且有仔细阅读了多个教程并搜索了论坛，但我仍然不知道如何继续。到目前为止，我已经得到了将文本转换为字符串格式然后过滤掉条目的序列文件

这是文件中一行的示例数据条目

Bob,123,USA,Pizza,Soda,,Blue
Bob,456,UK,Chocolate,Cheese,Soda,Green
Bob,12,USA,Chocolate,Pizza,Soda,Yellow
Mary,68,USA,Chips,Pasta,Chocolate,Blue

因此输出应该是元组 (Bob, Soda)，因为 soda 在 Bob 的条目中出现的次数最多。

import org.apache.hadoop.io._

var lines  = sc.sequenceFile("path",classOf[LongWritable],classOf[Text]).values.map(x => x.toString())
// converted to string since I could not get filter to run on Text and removing the longwritable

var filtered = lines.filter(_.split(",")(0) == "Bob");
// removed entries with all other users

var f_tuples = filtered.map(line => lines.split(",");
// split all the values

var f_simple = filtered.map(line => (line(0), (line(3), line(4), line(5))
// removed unnecessary fields

我现在遇到的这个问题是，我认为我有这个 [<name,[f,f,f]>] 结构，但我真的不知道如何继续将其展平并获得最受欢迎的食物。我需要组合所有条目，以便我有一个带有 a 的条目，然后获取值中最常见的元素。任何帮助，将不胜感激。谢谢

我尝试这样做是为了让它变平，但似乎我尝试得越多，数据结构就越复杂。

var f_trial = fpairs.groupBy(_._1).mapValues(_.map(_._2))
// the resulting structure was of type org.apache.spark.rdd.RDD[(String, Interable[(String, String, String)]

这是 f_trial 之后记录的 println 的样子

("Bob", List((Pizza, Soda,), (Chocolate, Cheese, Soda), (Chocolate, Pizza, Soda)))

括号分解

("Bob", 

List(

(Pizza, Soda, <missing value>),

(Chocolate, Cheese, Soda),

(Chocolate, Pizza, Soda)

) // ends List paren

) // ends first paren

Answer 1

我有时间。设置:

    import org.apache.spark.SparkContext
    import org.apache.spark.SparkContext._
    import org.apache.spark.SparkConf

    val conf = new SparkConf().setAppName("spark-scratch").setMaster("local")
    val sc = new SparkContext(conf)

    val data = """   
  Bob,123,USA,Pizza,Soda,,Blue
  Bob,456,UK,Chocolate,Cheese,Soda,Green
  Bob,12,USA,Chocolate,Pizza,Soda,Yellow
  Mary,68,USA,Chips,Pasta,Chocolate,Blue
  """.trim

    val records = sc.parallelize(data.split('\n'))

提取食物选择，并为每个选择一个元组 ((name, food), 1)

    val r2 = records.flatMap { r =>
      val Array(name, id, country, food1, food2, food3, color) = r.split(',');
      List(((name, food1), 1), ((name, food2), 1), ((name, food3), 1))
    }

将每个名称/食物组合加起来:

    val r3 = r2.reduceByKey((x, y) => x + y)

重新映射以便名称(仅)是键

    val r4 = r3.map { case ((name, food), total) => (name, (food, total)) }

在每一步选择计数最大的食物

    val res = r4.reduceByKey((x, y) => if (y._2 > x._2) y else x)

我们完成了

    println(res.collect().mkString)
    //(Mary,(Chips,1))(Bob,(Soda,3))

编辑:要收集一个人的所有最高计数的食物，我们只需更改最后两行:

从项目列表开始，总计:

val r5 = r3.map { case ((name, food), total) => (name, (List(food), total)) }

在相同的情况下，将食品列表与该分数连接起来

val res2 = r5.reduceByKey((x, y) => if (y._2 > x._2) y 
                                    else if (y._2 < x._2) x
                                    else (y._1:::x._1, y._2))

//(Mary,(List(Chocolate, Pasta, Chips),1))
//(Bob,(List(Soda),3))

例如，如果您想要前 3 名，则使用 aggregateByKey 来组装每个人最喜欢的食物列表，而不是第二个 reduceByKey