Spark RDD 分区效果

Question

我对重新分区操作感到困惑。请看下面的代码

import org.apache.spark._
import org.apache.log4j._


object FriendsByAge {

  def parseLine(line: String)={
    val fields = line.split(",")
    val age = fields(2).toInt
    val numFriends = fields(3).toInt

    (age, numFriends)
  }

  def main(args: Array[String]) = {

    Logger.getLogger("org").setLevel(Level.ERROR)

    val sc = new SparkContext("local[*]", "FriendsByAge")

    val lines = sc.textFile("./data/fakefriends-noheader.csv").repartition(1000)
    val rdd = lines.map(parseLine)

    println(rdd.getNumPartitions)

    val totalsByAge = rdd.mapValues(x=> (x,1)).reduceByKey((x, y) => (x._1+y._1, x._2 + y._2))

    println(totalsByAge.getNumPartitions)

    val averagesByAges = totalsByAge.mapValues(x => x._1/x._2)

    println(averagesByAges.getNumPartitions)
    val results = averagesByAges.collect()

    results.sortWith(_._2> _._2).foreach(println)
  }


}

这里我是将rdd读入1000个分区后重新分区。由于 map 操作会创建新的 RDD，并且不会保留分区。我仍然看到相同数量的分区。

问题是我如何知道子 RDD 是否会保留父 RDD 分区？子 RDD 使重新分区无效的条件是什么。

Answer 1

mapValues 不会改变已经生效的分区，它是一个 narrow 转换。你有两个。

reduceByKey 是关联的。 Spark 在本地聚合并将这些结果发送到驱动程序或相关分区 - 在您的情况下。如果你不为 number of partitions 使用 reduceByKey 上的参数，你会为新的 RDD 保留相同数量的分区，尽管分布不同。

Spark RDD 分区效果

Spark RDD Partition effects

partitioning

apache-spark

rdd