使用分区通过分割和有效地计算RDD组的关键

Question

我已经实现了一个解决方案，通过键分组RDD[K， V]，并根据每个组（K， RDD[V]）计算数据，使用分区按和分区器。然而，我不确定它是否真的有效，我想听听你的观点。

下面是一个示例案例：根据[K:Int，V:Int]列表，计算K每组的Vs平均值，知道它应该分布，并且V值可能非常大。这应该是：

List[K, V] => (K, mean(V))

简单的分区器类：

class MyPartitioner(maxKey: Int) extends Partitioner {

    def numPartitions = maxKey

    def getPartition(key: Any): Int = key match {
      case i: Int if i < maxKey => i
    }
  }

分区代码：

val l = List((1, 1), (1, 8), (1, 30), (2, 4), (2, 5), (3, 7))

      val rdd = sc.parallelize(l)
      val p =  rdd.partitionBy(new MyPartitioner(4)).cache()

      p.foreachPartition(x => {
        try {
          val r = sc.parallelize(x.toList)
          val id = r.first() //get the K partition id
          val v = r.map(x => x._2)
          println(id._1 + "->" + mean(v))
        } catch {
          case e: UnsupportedOperationException => 0
        }
      })

输出为：

1-

我的问题是：

1. 调用分区时到底发生了什么？（抱歉，我没有找到足够的规格）
2. 分区映射真的有效吗，知道在我的生产案例中，不会有太多的键（50个样本）和很多值（100万个样本）
3. 什么是paralellize（x.toList）的成本？这样做是一致的吗？（我需要一个RDD在输入的均值（））
4. 你自己会怎么做？

问候




Answer 1

您的代码不应该工作。不能将SparkContext对象传递给执行程序。（它不是Serializable。）我也不明白你为什么需要。

要计算平均值，你需要计算总和和和计数，并取它们的比率。默认的分区程序就可以了。

def meanByKey(rdd: RDD[(Int, Int)]): RDD[(Int, Double)] = {
  case class SumCount(sum: Double, count: Double)
  val sumCounts = rdd.aggregateByKey(SumCount(0.0, 0.0))(
    (sc, v) => SumCount(sc.sum + v, sc.count + 1.0),
    (sc1, sc2) => SumCount(sc1.sum + sc2.sum, sc1.count + sc2.count))
  sumCounts.map(sc => sc.sum / sc.count)
}

这是一个有效的单程计算，具有很好的通用性。