列表中的 Spark scala 过滤器元组

Question

我有一个如下所示的 Rdd

val m = sc.parallelize(Seq(("a",("x",1)), ("a",("y",2)), ("a",("z",2)), ("b",("x",1)),("b",("y",2))))

我通过使用如下所示的 groupByKey 转换了上面的 Rdd

val b = m.groupByKey.mapValues( _.toList)

结果：

(a,List((x,1), (y,2), (z,2)))
(b,List((x,1), (y,2)))

现在，我想过滤每个列表中具有最大值的元组所以预期的结果是

(a,List((y,2), (z,2)))
(b,List((y,2)))

Answer 1

考虑给定的序列是： val m = Seq(("a",("x",1)), ("a",("y",2)), ("a",("z",2)), ("b",("x",1)),("b",("y",2)))

val r1 = 
  m.groupBy(_._1)
   .map { case (k, v) => k -> v.map(_._2) }
   .map { case (k, v) => 
     k -> { 
       val sorted = v.sortWith { case (x, y) => x._2 > y._2 }
       val max = sorted.head._2

       sorted.takeWhile(_._2 == max) 
     }
   }
   .toList

结果如下： r1: List[(String, Seq[(String, Int)])] = List((b,List((y,2))), (a,List((y,2), (z,2))))

列表中的 Spark scala 过滤器元组

Spark scala filter tuples in a list

scala

apache-spark

rdd