In che modo Spark decide come partizionare un RDD?

Question

Supponiamo che io creo un tale RDD (sto usando Pyspark):

list_rdd = sc.parallelize(xrange(0, 20, 2), 6) 

poi stampare gli elementi partizionato con il metodo glom() e ottenere

[[0], [2, 4], [6, 8], [10], [12, 14], [16, 18]] 

Come ha Spark deciso come partizionare la mia lista? Da dove viene quella specifica scelta degli elementi? Avrebbe potuto accoppiarli in modo diverso, lasciando solo alcuni elementi diversi da 0 e 10, per creare le sei partizioni richieste. In una seconda corsa, le partizioni sono le stesse.

Utilizzando una gamma più ampia, con 29 elementi, ottengo partizioni della configurazione degli 2 elementi seguita da tre elementi:

list_rdd = sc.parallelize(xrange(0, 30, 2), 6) 
[[0, 2], [4, 6, 8], [10, 12], [14, 16, 18], [20, 22], [24, 26, 28]] 

Utilizzando una gamma minore di 9 elementi ottengo

list_rdd = sc.parallelize(xrange(0, 10, 2), 6) 
[[], [0], [2], [4], [6], [8]] 

Quindi quello che deduco è che Spark sta generando le partizioni suddividendo l'elenco in una configurazione dove il più piccolo possibile è seguito da raccolte più grandi e ripetuto.

La domanda è se c'è una ragione dietro questa scelta, che è molto elegante, ma fornisce anche vantaggi prestazionali?

Answer 1

Se non si specifica un partizionatore specifico, questo è "casuale" in quanto dipende dall'implementazione specifica di tale RDD. In questo caso puoi andare allo ParallelCollectionsRDD per approfondire ulteriormente.

getPartitions è definito come:

val slices = ParallelCollectionRDD.slice(data, numSlices).toArray 
slices.indices.map(i => new ParallelCollectionPartition(id, i, slices(i))).toArray 

dove slice è commentato come (riformattato per adattarsi meglio):

/** 
* Slice a collection into numSlices sub-collections. 
* One extra thing we do here is to treat Range collections specially, 
* encoding the slices as other Ranges to minimize memory cost. 
* This makes it efficient to run Spark over RDDs representing large sets of numbers. 
* And if the collection is an inclusive Range, 
* we use inclusive range for the last slice. 
*/ 

noti che ci sono alcune considerazioni per quanto riguarda la memoria. Quindi, di nuovo, questo sarà specifico per l'implementazione.