Metoda parallelize() w SparkContext

Question

Próbuję zrozumieć wpływ podania innej numSlices na metodę parallelize() w SparkContext. Podane poniżej jest Syntax metody

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

wpadłem zapłonie powłokę w trybie lokalnym

spark-shell --master local 

Moje zrozumienie jest, numSlices zdecyduje, że żaden z partycji wypadkowej RDD (po wywołaniu sc.parallelize()). Rozważyć kilka przykładów poniżej

Case 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 

Case 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 

Case 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 

Case 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 

Pytanie 1 W przypadku 3 & Przypadek 4, spodziewałem rozmiar partycji być 3 & 4 odpowiednio, ale w obu przypadkach mają rozmiar partycji tylko 2. Jaki jest tego powód?

Pytanie 2: W każdym przypadku istnieje numer powiązany z ParallelCollectionRDD[no]. W przypadku 1 jest to ParallelCollectionRDD[0], w przypadku 2 jest to ParallelCollectionRDD[1] & tak dalej. Co dokładnie oznaczają te liczby?

Answer 1

Pytanie 1: To literówka z Twojej strony. Nazywasz res3.partitions.size, zamiast res5 i res7 odpowiednio. Kiedy robię to z poprawną liczbą, działa zgodnie z oczekiwaniami.

Pytanie 2: To jest identyfikator RDD w Spark Context, używany do utrzymywania prostego wykresu. Zobacz, co się stanie, gdy trzykrotnie wykonam to samo polecenie:

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 

Dostępne są teraz trzy różne RDD z trzema różnymi identyfikatorami. Możemy sprawdzić, czy:

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