Capire il caching di Spark

Question

Sto cercando di capire come funziona il cache di Spark.

Qui è la mia comprensione ingenuo, per favore fatemelo sapere se mi manca qualcosa:

val rdd1 = sc.textFile("some data") 
rdd1.cache() //marks rdd1 as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

In quanto sopra, rdd1 sarà caricato da disco (ad esempio HDFS) solo una volta. (quando rdd2 viene salvato presumo) e poi dalla cache (supponendo che ci sia abbastanza RAM) quando viene salvato rdd3)

Ora ecco la mia domanda. Diciamo che voglio memorizzare nella cache rdd2 e rdd3 poiché saranno entrambi utilizzati in seguito, ma non ho bisogno di rdd1 dopo averli creati.

Fondamentalmente c'è la duplicazione, non è vero? Poiché una volta calcolati rdd2 e rdd3, non ho più bisogno di rdd1, probabilmente dovrei eliminarlo, giusto? la domanda è quando?

Funzionerà? (Opzione A)

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 

Vuol scintilla aggiungere la chiamata unpersist al DAG? o è fatto immediatamente? se è fatto immediatamente, quindi fondamentalmente rdd1 non verrà memorizzato nella cache quando leggo da rdd2 e rdd3, giusto?

Devo farlo in questo modo invece (Opzione B)?

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 

rdd2.cache() 
rdd3.cache() 

rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

rdd1.unpersist() 

Quindi la domanda è questa: è l'opzione A abbastanza buono? Ad esempio, rdd1 caricherà ancora il file solo una volta? O devo andare con l'opzione B?

Answer 1

Sembrerebbe che sia richiesta l'opzione B. Il motivo è legato al modo in cui persistono/cache e unpersist vengono eseguiti da Spark. Poiché le trasformazioni RDD si limitano a creare descrizioni DAG senza esecuzione, nell'opzione A quando si chiama unpersist, si hanno solo descrizioni del lavoro e non un'esecuzione in esecuzione.

Ciò è rilevante in quanto una chiamata cache o persist aggiunge semplicemente l'RDD a una mappa di RDD contrassegnati come permanenti durante l'esecuzione del lavoro. Tuttavia, unpersist dice direttamente a blockManager di rimuovere l'RDD dalla memoria e rimuove il riferimento nella mappa degli RDD persistenti.

persist function

unpersist function

modo che avrebbe bisogno di chiamare unpersist dopo Spark effettivamente eseguito e memorizzato il RDD con il gestore del blocco.

I commenti per il RDD.persist metodo di suggerimento in questa direzione: rdd.persist

Answer 2

In opzione A, non hanno dimostrato quando si chiama l'azione (chiamata per salvare)

val rdd1 = sc.textFile("some data") 
rdd.cache() //marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

Se la sequenza è come sopra, l'opzione A dovrebbe usare la versione cache di rdd1 per calcolare sia rdd2 che rdd 3

Answer 3

L'opzione B è un approccio ottimale con piccoli ritocchi. Utilizzare metodi di azione meno costosi. Nell'approccio menzionato dal tuo codice, saveAsTextFile è un'operazione costosa, sostituiscila con il metodo count.

idea è quella di rimuovere il grande rdd1 da DAG, se non è rilevante per l'ulteriore calcolo (dopo rdd2 e rdd3 sono creati)

approccio Aggiornato dal codice

val rdd1 = sc.textFile("some data").cache() 
val rdd2 = rdd1.filter(...).cache() 
val rdd3 = rdd1.map(...).cache() 

rdd2.count 
rdd3.count 

rdd1.unpersist()