Spark: differenza di semantica tra riduzione e riduzione. KEY

Question

Nella documentazione di Spark, si afferma che il metodo RDDs reduce richiede una funzione binaria associativa AND commutativa.

Tuttavia, il metodo reduceByKey richiede SOLO una funzione binaria associativa.

sc.textFile("file4kB", 4) 

Ho fatto alcuni test e apparentemente è il comportamento che ottengo. Perché questa differenza? Perché reduceByKey assicura che la funzione binaria sia sempre applicata in un determinato ordine (per ovviare alla mancanza di commutatività) quando lo reduce non lo fa?

esempio, se un carico qualche (piccolo) testo con 4 partizioni (minimo):

val r = sc.textFile("file4k", 4) 

quindi:

r.reduce(_ + _) 

restituisce una stringa in cui parti non sono sempre nello stesso ordine, mentre:

r.map(x => (1,x)).reduceByKey(_ + _).first 

restituisce sempre la stessa stringa (dove tutto è nello stesso ordine rispetto all'originale) l file).

(Ho controllato con r.glom e il contenuto del file è effettivamente distribuito su 4 partizioni, non c'è nessuna partizione vuota).

Answer 1

Per quanto mi riguarda, si tratta di un errore nella documentazione e i risultati che si vedono sono semplicemente accidentali. La pratica, other resources e un semplice analysis of the code mostrano che la funzione passata a reduceByKey dovrebbe essere non solo associativa ma anche commutativa.

• pratica - mentre sembra che l'ordine è conservato in una modalità locale non è più vero quando si esegue Spark in un cluster, compresa la modalità stand-alone.

• altre risorse - per citare Data Exploration Using Spark da AmpCamp 3:

  C'è un metodo conveniente chiamato reduceByKey in Spark proprio per questo motivo. Si noti che il secondo argomento per ridurreByKey determina il numero di riduttori da utilizzare. Per impostazione predefinita, Spark presuppone che la funzione di riduzione sia commutativa e associativa e applica i combinatori sul lato del mapper. Codice

• - reduceByKey è implementato utilizzando combineByKeyWithClassTag e crea ShuffledRDD. Poiché Spark non garantisce l'ordine dopo la mischia, l'unico modo per ripristinarlo sarebbe quello di allegare alcuni metadati ai record parzialmente ridotti. Per quanto posso dire non succede niente del genere.

Su un lato nota reduce come è implementato in PySpark funzionano bene con una funzione che è commutativo soltanto. Ovviamente è solo un dettaglio di un'implementazione e non una parte del contratto.

Answer 2

Secondo la documentazione del codice, aggiornato/corretto di recente.(grazie @ zero323):

reduceByKey unisce i valori per ogni chiave utilizzando una funzione di riduzione associativa e commutativa. Questo eseguirà anche l'unione localmente su ogni mapper prima di inviare i risultati a un riduttore, analogamente a un "combinatore" in MapReduce.

Quindi in realtà si trattava di un errore di documentazione come @ zero323 evidenziato nella sua risposta.

È possibile controllare i seguenti link per il codice per assicurarsi che:

• https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala#L304

• https://github.com/apache/spark/blob/master/python/pyspark/rdd.py#L1560