SparkContext中的并行化()方法

Question

我试图理解将不同的numSlices赋予SparkContext中的parallelize()方法的效果。给出的是方法的Syntax

def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)
(implicit arg0: ClassTag[T]): RDD[T]

我在本地模式中运行了火花壳。

spark-shell --master local

我的理解是，numSlices决定结果的RDD的分区的no (在调用sc.parallelize()之后)。考虑下面的几个例子

案例1

scala> sc.parallelize(1 to 9, 1);
res0: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:22
scala> res0.partitions.size
res2: Int = 1

案例2

scala> sc.parallelize(1 to 9, 2);
res3: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:22
scala> res3.partitions.size
res4: Int = 2

案例3

scala> sc.parallelize(1 to 9, 3);
res5: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at <console>:22
scala> res3.partitions.size
res6: Int = 2

案例4

scala> sc.parallelize(1 to 9, 4);
res7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at <console>:22
scala> res3.partitions.size
res8: Int = 2

问题1：在case 3 & case 4中，我期望分区大小分别为3和4，但这两种情况的分区大小都只有2。原因是什么？

问题2：在每种情况下都有一个与ParallelCollectionRDD[no]相关联的数字。在第1种情况下是ParallelCollectionRDD[0]，在第2种情况下是ParallelCollectionRDD[1] &以此类推。这些数字到底意味着什么？

Answer 1

问题1：这是您的错误。您正在调用res3.partitions.size，而不是分别调用res5和res7。当我用正确的数字做这件事时，它会像预期的那样工作。

问题2：这是火花上下文中RDD的id，用于保持图形的直线。看看当我运行三次相同的命令时会发生什么：

scala> sc.parallelize(1 to 9,1)
res0: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:22

scala> sc.parallelize(1 to 9,1)
res1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:22

scala> sc.parallelize(1 to 9,1)
res2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at <console>:22

现在有三个不同的RDDs，具有三个不同的ids。我们可以运行以下命令来检查：

scala> (res0.id, res1.id, res2.id)
res3: (Int, Int, Int) = (0,1,2)