spark3.0教程：sc.makeRDD()并行度和分区方式

作者：马育民 • • 阅读：10316

说明

默认情况下，Spark 可以将一个作业切分多个任务后，发送给 Executor 节点并行计算

并行度： 指并行计算的任务数量。这个数量可以在构建 RDD 时指定。

有几个并行任务，就有几个分区

通过 makeRDD() 方式生成的rdd

val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
rdd.saveAsTextFile("res")

sc.stop()

解释：

  • 如果设置 local[*] ，表示本地运行，并行度默认为核心数量

  • 如果设置 local ，并行度默认为1

  • sc.makeRDD()底层调用 sc.parallelize()

设置并行度数量

方式1

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),5)

方式2

sparConf.set("spark.default.parallelism","5");

验证

val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
//    sparConf.set("spark.default.parallelism","3");
val sc = new SparkContext(sparConf)

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),3)
rdd.saveAsTextFile("res")

sc.stop()

分区方式

下面代码中，

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),3)

是如何将 List(1,2,3,4) 分成3个部分的？

通过源码可知： ParallelCollectionRDD.getPartitions() -> ParallelCollectionRDD.slice(data, numSlices) 方法中（第127行代码）

关键代码如下：

def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
  (0 until numSlices).iterator.map { i =>
    val start = ((i * length) / numSlices).toInt
    val end = (((i + 1) * length) / numSlices).toInt
    (start, end)
  }
}

已知：

  • length：是seq的长度

  • numSlices：是分区数量

计算：

  • val start = ((i * length) / numSlices).toInt

  • val end = (((i + 1) * length) / numSlices).toInt

计算方式：

以：sc.makeRDD(List(1,2,3,4),3)为例，

  1. 第一次运算，i=0，length=4，计算：start=0，end=1，截取索引是0的元素，即：第一个文件里只有1
  2. 第二次运算，i=1，length=4，计算：start=1，end=2，截取索引是1的元素，即：第一个文件里只有2
  3. 第三次运算，i=2，length=4，计算：start=2，end=4，截取索引是2,3的元素，即：第一个文件里有3,4

通过 parallelize() 方式生成的rdd

sc.makeRDD() 底层调用 sc.parallelize()

sc.parallelize(data)

原文出处：/show_1IX1Inv4d9vX.html

吉ICP备15003043号