paul和mattinbits提供的解决方案将数据洗牌两次—一次按名称和食物执行reduce，一次按名称执行reduce。只用一次洗牌就可以解决这个问题。

/**Generate key-food_count pairs from a splitted line**/
def bitsToKeyMapPair(xs: Array[String]): (String, Map[String, Long]) = {
  val key = xs(0)
  val map = xs
    .drop(3) // Drop name..country
    .take(3) // Take food
    .filter(_.trim.size !=0) // Ignore empty
    .map((_, 1L)) // Generate k-v pairs
    .toMap // Convert to Map
    .withDefaultValue(0L) // Set default

  (key, map)
}

/**Combine two count maps**/
def combine(m1: Map[String, Long], m2: Map[String, Long]): Map[String, Long] = {
  (m1.keys ++ m2.keys).map(k => (k, m1(k) + m2(k))).toMap.withDefaultValue(0L)
}

val n: Int = ??? // Number of favorite per user

val records = lines.map(line => bitsToKeyMapPair(line.split(",")))
records.reduceByKey(combine).mapValues(_.toSeq.sortBy(-_._2).take(n))

如果你不是一个纯粹主义者，你可以取代 scala.collection.immutable.Map 与 scala.collection.mutable.Map 进一步提高绩效。

下面是一个完整的示例：

import org.apache.spark.{SparkContext, SparkConf}

object Main extends App {

  val data = List(
    "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")

  val sparkConf = new SparkConf().setMaster("local").setAppName("example")
  val sc = new SparkContext(sparkConf)

  val lineRDD = sc.parallelize(data)

  val pairedRDD = lineRDD.map { line =>
    val fields = line.split(",")
    (fields(0), List(fields(3), fields(4), fields(5)).filter(_ != ""))
  }.filter(_._1 == "Bob")

  /*pairedRDD.collect().foreach(println)
    (Bob,List(Pizza, Soda))
    (Bob,List(Chocolate, Cheese, Soda))
    (Bob,List(Chocolate, Pizza, Soda))
   */

  val flatPairsRDD = pairedRDD.flatMap {
    case (name, foodList) => foodList.map(food => ((name, food), 1))
  }

  /*flatPairsRDD.collect().foreach(println)
    ((Bob,Pizza),1)
    ((Bob,Soda),1)
    ((Bob,Chocolate),1)
    ((Bob,Cheese),1)
    ((Bob,Soda),1)
    ((Bob,Chocolate),1)
    ((Bob,Pizza),1)
    ((Bob,Soda),1)
   */

  val nameFoodSumRDD = flatPairsRDD.reduceByKey((a, b) => a + b)

  /*nameFoodSumRDD.collect().foreach(println)
    ((Bob,Cheese),1)
    ((Bob,Soda),3)
    ((Bob,Pizza),2)
    ((Bob,Chocolate),2)
   */

  val resultsRDD = nameFoodSumRDD.map{
    case ((name, food), count) => (name, (food,count))
  }.groupByKey.map{
    case (name, foodCountList) => (name, foodCountList.toList.sortBy(_._2).reverse.head)
  }

  resultsRDD.collect().foreach(println)
  /*
      (Bob,(Soda,3))
   */

  sc.stop()
}

我找到时间了。设置：

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)

重新Map，使名称（仅）成为键

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))

如果你想要前三名，那么就用 `aggregateByKey` 列出每个人最喜欢的食物而不是第二种 `reduceByKey`