Spark RDD - evitare shuffle - Il partizionamento aiuta a elaborare file enormi?

Question

Ho un'applicazione con circa 10 file flat ciascuno del valore di oltre 200 MM + record in essi. La logica aziendale implica l'unione di tutti loro in modo sequenziale.

mio ambiente: 1 master - 3 slave (per i test ho assegnato un 1GB di memoria per ciascun nodo)

La maggior parte del codice appena fa il seguito per ogni join

RDD1 = sc.textFile(file1).mapToPair(..) 

RDD2 = sc.textFile(file2).mapToPair(..) 

join = RDD1.join(RDD2).map(peopleObject) 

Ogni suggerimento per la messa a punto, come ripartizionare, parallelizzare ..? In caso affermativo, le migliori pratiche in venire con un buon numero per il ripartizionamento?

con la configurazione corrente del lavoro prende più di un'ora e vedo la scrittura casuale per quasi tutti i file è> 3 GB

Answer 1

Se siamo sempre associarsi ad uno RDD (diciamo rdd1) con tutti gli altri, l'idea è partizionare quel RDD e poi persisterlo.

Ecco implementazione sudo-Scala (può essere facilmente convertito in Python o Java):

val rdd1 = sc.textFile(file1).mapToPair(..).partitionBy(new HashPartitioner(200)).cache() 

Fino a qui abbiamo rdd1 da hash in 200 partizioni. La prima volta che verrà valutata verrà mantenuta (memorizzata nella cache).

Ora leggiamo altri due rdds e li uniamo.

val rdd2 = sc.textFile(file2).mapToPair(..) 
val join1 = rdd1.join(rdd2).map(peopleObject) 
val rdd3 = sc.textFile(file3).mapToPair(..) 
val join2 = rdd1.join(rdd3).map(peopleObject) 

Si noti che per gli RDD di ricontrazione non li partizioniamo e non li memorizziamo nella cache.

Spark vedrà che rdd1 è già una partizione con hash e utilizzerà le stesse partizioni per tutti i join rimanenti. Quindi rdd2 e rdd3 mescoleranno le loro chiavi alle stesse posizioni in cui si trovano le chiavi di rdd1.

Per rendere più chiaro, supponiamo di non eseguire la partizione e di utilizzare lo stesso codice mostrato dalla domanda; Ogni volta che facciamo un join entrambi i dischi verranno mescolati. Ciò significa che se abbiamo N join per rdd1, la versione non partizionata rimescola rdd1 N volte. L'approccio partizionato riprodurrà rdd1 una sola volta.

Answer 2

In pratica, con dataset di grandi dimensioni (5, 100G + ciascuno), ho visto che il join funziona meglio quando si co-partizionano tutti gli RDD coinvolti in una serie di join prima di iniziare a unirli.

RDD1 = sc.textFile(file1).mapToPair(..).partitionBy(new HashPartitioner(2048)) 

RDD2 = sc.textFile(file2).mapToPair(..).partitionBy(new HashPartitioner(2048)) 
. 
. 
. 
RDDN = sc.textFile(fileN).mapToPair(..).partitionBy(new HashPartitioner(2048)) 

//start joins 

RDD1.join(RDD2)...join(RDDN)

---

Nota a margine: Mi riferisco a questo tipo di aderire come un albero di join (ogni RDD un tempo utilizzato). La logica è presentato nella seguente bella foto tratte da Spark-UI: