scala - usa - particionar disco duro en ubuntu

Chispa: ¿aumentar el número de particiones sin causar una confusión?

Question

Cuando se reduce el número de particiones, se puede usar la coalesce , lo cual es genial porque no causa una confusión y parece funcionar instantáneamente (no requiere una etapa de trabajo adicional).

Me gustaría hacer lo contrario a veces, pero la repartition induce una confusión. Creo que hace unos meses conseguí que esto funcionara al usar CoalescedRDD con balanceSlack = 1.0 , entonces lo que sucedería es que dividiría una partición para que las particiones resultantes estuvieran ubicadas en el mismo nodo (IO de red tan pequeña).

Este tipo de funcionalidad es automática en Hadoop, uno simplemente ajusta el tamaño de la división. No parece funcionar de esta manera en Spark a menos que uno esté disminuyendo el número de particiones. Creo que la solución podría ser escribir un particionador personalizado junto con un RDD personalizado en el que definamos getPreferredLocations ... ¿pero creo que es algo tan simple y común de hacer que debe haber una forma directa de hacerlo?

Las cosas intentaron:

.set("spark.default.parallelism", partitions) en mi SparkConf , y cuando en el contexto de leer parquet he intentado sqlContext.sql("set spark.sql.shuffle.partitions= ... , que en 1.0 .0 causa un error Y realmente no quiero lo que quiero, quiero que el número de partición cambie en todos los tipos de trabajo, no solo en orden aleatorio.

Answer 1

Como usted sabe, pyspark usa algún tipo de forma "perezosa" de correr. Solo realizará el cálculo cuando haya alguna acción que hacer (por ejemplo, "df.count ()" o "df.show ()". Entonces, lo que puede hacer es definir una partición aleatoria entre esas acciones.

Puedes escribir :

sparkSession.sqlContext().sql("set spark.sql.shuffle.partitions=100")
# you spark code here with some transformation and at least one action
df = df.withColumn("sum", sum(df.A).over(your_window_function))
df.count() # your action
df = df.filter(df.B <10)
df = df.count()
sparkSession.sqlContext().sql("set spark.sql.shuffle.partitions=10")
# you reduce the number of partition because you know you will have a lot
# less data
df = df.withColumn("max", max(df.A).over(your_other_window_function))
df.count() # your action

Answer 2

Mira este espacio

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

Este tipo de característica obvia realmente simple se implementará eventualmente, supongo que justo después de que terminen todas las características innecesarias en los Dataset de Dataset .

Answer 3

No entiendo exactamente cuál es tu punto. ¿Quiere decir que ahora tiene 5 particiones, pero después de la próxima operación desea que los datos se distribuyan a 10? Porque tener 10, pero seguir usando 5 no tiene mucho sentido ... El proceso de enviar datos a nuevas particiones tiene que suceder en algún momento.

Al hacer la coalesce , puede deshacerse de las particiones no guardadas, por ejemplo: si inicialmente tenía 100, pero luego de reducirByKey obtuvo 10 (ya que solo había 10 teclas), puede configurar la coalesce .

Si desea que el proceso vaya por otro lado, podría forzar algún tipo de partición:

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

No estoy seguro de que es lo que estás buscando, pero espero que sí.