如何在scala中传递自定义函数来减少RDD的ByKey

Question

我的要求是找到RDD中每个组的最大值。

我尝试了以下方法;

scala> val x = sc.parallelize(Array(Array("A",3), Array("B",5), Array("A",6)))
x: org.apache.spark.rdd.RDD[Array[Any]] = ParallelCollectionRDD[0] at parallelize at <console>:27
scala> x.collect
res0: Array[Array[Any]] = Array(Array(A, 3), Array(B, 5), Array(A, 6))          
scala> x.filter(math.max(_,_))
<console>:30: error: wrong number of parameters; expected = 1
              x.filter(math.max(_,_))
                               ^

我也尝试了以下方法;选项 1：

scala> x.filter((x: Int, y: Int) => { math.max(x,y)} )
<console>:30: error: type mismatch;
 found   : (Int, Int) => Int
 required: Array[Any] => Boolean
              x.filter((x: Int, y: Int) => { math.max(x,y)} )

选项 2：

scala> val myMaxFunc = (x: Int, y: Int) => { math.max(x,y)}
myMaxFunc: (Int, Int) => Int = <function2>
scala> myMaxFunc(56,12)
res10: Int = 56
scala> x.filter(myMaxFunc(_,_) )
<console>:32: error: wrong number of parameters; expected = 1
              x.filter(myMaxFunc(_,_) )

如何做到这一点？

Answer 1

我只能猜测，但你可能想这样做：

val rdd = sc.parallelize(Array(("A", 3), ("B", 5), ("A", 6)))
val max = rdd.reduceByKey(math.max)
println(max.collect().toList)  // List((B,5), (A,6))

而不是"如何做到这一点？"你应该解释你的预期结果是什么。我认为你犯了一些错误：

• 使用filter而不是reduceByKey(为什么？？
• reduceByKey仅适用于PairRDD，因此您需要元组而不是Array[Any]元组(无论如何这是一种糟糕的类型(
• 你不需要为math.max编写自己的包装函数，你可以按原样使用