apache-spark - started - spark todf




Spark-repartition() vs coalesce() (8)

De acuerdo con Learning Spark

Tenga en cuenta que repartir sus datos es una operación bastante costosa. Spark también tiene una versión optimizada de repartition () llamada coalesce () que permite evitar el movimiento de datos, pero solo si está disminuyendo el número de particiones RDD.

Una diferencia que obtengo es que con repartición () el número de particiones se puede aumentar / disminuir, pero con fusión () el número de particiones solo se puede disminuir.

Si las particiones se distribuyen en varias máquinas y se ejecuta coalesce (), ¿cómo puede evitar el movimiento de datos?


A todas las excelentes respuestas, me gustaría agregar que volver a particionar es una de las mejores opciones para aprovechar la paralelización de datos y la fusión ofrece una opción económica para reducir la partición y muy útil al escribir datos en HDFS o algún otro receptor para aprovechar Grandes escrituras. He encontrado esto útil cuando escribo datos en formato parquet para aprovechar al máximo.


De una manera simple COALESCE: - es solo para disminuir el no de particiones, sin mezclar datos, solo comprime las particiones

REPARACIÓN: es tanto para aumentar como para disminuir el número de particiones, pero se baraja

Ejemplo:-

val rdd = sc.textFile("path",7)
rdd.repartition(10)
rdd.repartition(2)

Ambos funcionan bien

Pero generalmente vamos por estas dos cosas cuando necesitamos ver la salida en un clúster, vamos con esto.


La respuesta de Justin es asombrosa y esta respuesta se profundiza.

El algoritmo de repartition realiza una combinación completa y crea nuevas particiones con datos que se distribuyen de manera uniforme. Creemos un DataFrame con los números del 1 al 12.

val x = (1 to 12).toList
val numbersDf = x.toDF("number")

numbersDf contiene 4 particiones en mi máquina.

numbersDf.rdd.partitions.size // => 4

Así es como se dividen los datos en las particiones:

Partition 00000: 1, 2, 3
Partition 00001: 4, 5, 6
Partition 00002: 7, 8, 9
Partition 00003: 10, 11, 12

Hagamos una combinación completa con el método de repartition y obtengamos estos datos en dos nodos.

val numbersDfR = numbersDf.repartition(2)

Así es como se numbersDfR datos numbersDfR en mi máquina:

Partition A: 1, 3, 4, 6, 7, 9, 10, 12
Partition B: 2, 5, 8, 11

El método de repartition nuevas particiones y distribuye de manera uniforme los datos en las nuevas particiones (la distribución de datos es más uniforme incluso para conjuntos de datos más grandes).

Diferencia entre coalesce y repartition

coalesce usa particiones existentes para minimizar la cantidad de datos que se barajan. repartition crea nuevas particiones y baraja completamente. coalesce da como resultado particiones con diferentes cantidades de datos (a veces particiones que tienen tamaños muy diferentes) y la repartition da como resultado particiones de aproximadamente el mismo tamaño.

¿La coalesce o repartition más rápida?

coalesce puede ejecutarse más rápido que la repartition , pero las particiones de tamaño desigual generalmente son más lentas para trabajar que las particiones de igual tamaño. Por lo general, deberá volver a particionar los conjuntos de datos después de filtrar un conjunto de datos grande. He descubierto que la repartition es más rápida en general porque Spark está diseñado para funcionar con particiones de igual tamaño.

Lea esta publicación de blog si desea aún más detalles.


Lo que se deduce del code y los documentos de código es que la coalesce(n) es lo mismo que la coalesce(n, shuffle = false) y la repartition(n) es lo mismo que la coalesce(n, shuffle = true)

Por lo tanto, tanto la coalesce como la repartition pueden usarse para aumentar el número de particiones

Con shuffle = true, en realidad puede unirse a un mayor número de particiones. Esto es útil si tiene una pequeña cantidad de particiones, digamos 100, potencialmente con unas pocas particiones que son anormalmente grandes.

Otra nota importante para acentuar es que si disminuye drásticamente el número de particiones, debería considerar usar una versión coalesce de coalesce (igual que la repartition en ese caso). Esto permitirá que sus cálculos se realicen en paralelo en particiones principales (tarea múltiple).

Sin embargo, si está haciendo una fusión drástica, por ejemplo, a numPartitions = 1, esto puede hacer que su cálculo se realice en menos nodos de los que desee (por ejemplo, un nodo en el caso de numPartitions = 1). Para evitar esto, puede pasar shuffle = true. Esto agregará un paso aleatorio, pero significa que las particiones ascendentes actuales se ejecutarán en paralelo (según lo que sea la partición actual).

Consulte también la respuesta relacionada here


Para alguien que tuvo problemas para generar un solo archivo csv de PySpark (AWS EMR) como salida y guardarlo en s3, el uso de la partición ayudó. La razón es que la fusión no puede hacer una mezcla completa, pero la distribución sí. Esencialmente, puede aumentar o disminuir el número de particiones usando la división, pero solo puede disminuir el número de particiones (pero no 1) usando la fusión. Aquí está el código para cualquier persona que intente escribir un csv de AWS EMR en s3:

df.repartition(1).write.format('csv')\
.option("path", "s3a://my.bucket.name/location")\
.save(header = 'true')

Pero también debe asegurarse de que, si se trata de datos de gran tamaño, los datos que están llegando a los nodos de fusión deben estar altamente configurados. Debido a que todos los datos se cargarán en esos nodos, puede provocar una excepción de memoria. Aunque la reparación es costosa, prefiero usarla. Ya que baraja y distribuye los datos por igual.

Sea prudente para seleccionar entre fusión y reparto.


Un punto adicional a tener en cuenta aquí es que, como el principio básico de Spark RDD es la inmutabilidad. La repartición o fusión creará un nuevo RDD. El RDD base continuará teniendo existencia con su número original de particiones. En caso de que el caso de uso exija que el RDD persista en la memoria caché, entonces se debe hacer lo mismo para el RDD recién creado.

scala> pairMrkt.repartition(10)
res16: org.apache.spark.rdd.RDD[(String, Array[String])] =MapPartitionsRDD[11] at repartition at <console>:26

scala> res16.partitions.length
res17: Int = 10

scala>  pairMrkt.partitions.length
res20: Int = 2

repartition : se recomienda usar la partición mientras se aumenta el número de particiones, ya que implica la combinación aleatoria de todos los datos.

coalesce : se recomienda usar coalesce sin reducir las particiones. Por ejemplo, si tiene 3 particiones y desea reducirlas a 2 particiones, Coalesce moverá los datos de la tercera partición a las particiones 1 y 2. Las particiones 1 y 2 permanecerán en el mismo contenedor. entre ejecutor será alto e impacta el rendimiento.

El rendimiento inteligente coalesce mejor el rendimiento que el repartition mientras reduce el número de particiones.





rdd