Spark: aumentare il numero di partizioni senza causare un shuffle?

Question

Quando si riduce il numero di partizioni si può usare coalesce, che è ottimo perché non causa uno shuffle e sembra funzionare immediatamente (non richiede una fase di lavoro aggiuntiva).

Mi piacerebbe fare l'opposto a volte, ma repartition induce un shuffle. Penso che alcuni mesi fa ho effettivamente funzionato usando CoalescedRDD con balanceSlack = 1.0 - quindi cosa succederebbe sarebbe dividere una partizione in modo che la partizione risultante posizione in cui tutti sullo stesso nodo (così piccolo IO rete).

Questo tipo di funzionalità è automatico in Hadoop, basta modificare le dimensioni divise. Non sembra funzionare in questo modo in Spark a meno che non si stia riducendo il numero di partizioni. Penso che la soluzione potrebbe essere quella di scrivere un partizionatore personalizzato insieme a un RDD personalizzato in cui definiamo getPreferredLocations ... ma ho pensato che è una cosa così semplice e comune da fare sicuramente ci deve essere un modo semplice per farlo?

Cose cercato:

.set("spark.default.parallelism", partitions) sul mio SparkConf, e quando nel contesto della lettura parquet ho provato sqlContext.sql("set spark.sql.shuffle.partitions= ..., che sulla 1.0.0 provoca un errore e non vuole veramente che voglio, voglio partizione numero da cambiare tra tutti i tipi di lavoro, non solo shuffle.

Answer 1

Guarda questo spazio

https://issues.apache.org/jira/browse/SPARK-5997

Questo tipo di molto semplice caratteristica evidente finirà per essere attuate - credo che solo al termine della tutte le funzioni inutili in Dataset s.

Answer 2

Non capisco esattamente qual è il tuo punto. Vuoi dire che hai ora 5 partizioni, ma dopo l'operazione successiva vuoi i dati distribuiti a 10? Perché avere 10, ma usare ancora 5 non ha molto senso ... Il processo di invio dei dati a nuove partizioni deve avvenire qualche volta.

Quando si esegue coalesce, è possibile eliminare partizioni non desiderate, ad esempio: se inizialmente si era 100, ma dopo aver ridottoByKey si ottiene 10 (come là dove solo 10 chiavi), è possibile impostare coalesce.

Se si desidera che il processo di andare nella direzione opposta, si può solo imporre un qualche tipo di partizionamento:

[RDD].partitionBy(new HashPartitioner(100)) 

io non sono sicuro che è quello che stai cercando, ma spero così.