Nuestros mejores investigadores agotaron sus depósitos de café, por su búsqueda noche y día por la respuesta, hasta que Tobías halló la respuesta en Beanstalk por lo tanto hoy la comparte con nosotros.
Solución:
Evita un completo barajar. Si se sabe que el número está disminuyendo, entonces el ejecutor puede mantener los datos de forma segura en el número mínimo de particiones, solo moviendo los datos de los nodos adicionales a los nodos que mantuvimos.
Entonces, sería algo como esto:
Node 1 = 1,2,3
Node 2 = 4,5,6
Node 3 = 7,8,9
Node 4 = 10,11,12
Entonces coalesce hasta 2 particiones:
Node 1 = 1,2,3 + (10,11,12)
Node 3 = 7,8,9 + (4,5,6)
Observe que el Nodo 1 y el Nodo 3 no requerían que se movieran sus datos originales.
La respuesta de Justin es increíble y esta respuesta profundiza más.
El repartition El algoritmo hace una mezcla completa y crea nuevas particiones con datos que se distribuyen uniformemente. Vamos a crear 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 un barajado completo con el repartition método y obtenga estos datos en dos nodos.
val numbersDfR = numbersDf.repartition(2)
Así es como el numbersDfR los datos están particionados en mi máquina:
Partition A: 1, 3, 4, 6, 7, 9, 10, 12
Partition B: 2, 5, 8, 11
El repartition El método crea nuevas particiones y distribuye uniformemente los datos en las nuevas particiones (la distribución de datos es más uniforme para conjuntos de datos más grandes).
Diferencia entre coalesce y repartition
coalesce utiliza particiones existentes para minimizar la cantidad de datos que se mezclan. repartition crea nuevas particiones y hace una mezcla completa. coalesce da como resultado particiones con diferentes cantidades de datos (a veces particiones que tienen tamaños muy diferentes) y repartition da como resultado particiones de aproximadamente el mismo tamaño.
Es coalesce o repartition ¿más rápido?
coalesce puede correr más rápido que repartition, pero las particiones de tamaño desigual son generalmente 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 encontrado repartition para ser más rápido en general porque Spark está diseñado para funcionar con particiones del mismo tamaño.
NB: curiosamente he observado que la partición puede aumentar el tamaño de los datos en el disco. Asegúrese de ejecutar pruebas cuando utilice la partición o la fusión en grandes conjuntos de datos.
Lea esta publicación de blog si desea obtener aún más detalles.
Cuándo usará coalesce & repartition en la práctica
- Consulte esta pregunta sobre cómo usar coalesce & repartition para escribir un DataFrame en un solo archivo
- Es fundamental volver a particionar después de ejecutar consultas de filtrado. La cantidad de particiones no cambia después del filtrado, por lo que si no vuelve a particionar, tendrá demasiadas particiones de memoria (cuanto más reduzca el filtro el tamaño del conjunto de datos, mayor será el problema). Cuidado con el problema de la partición vacía.
- particiónBy se utiliza para escribir datos en particiones en el disco. Deberá usar repartition/coalesce para particionar sus datos en la memoria correctamente antes de usar particiónBy.
Un punto adicional a tener en cuenta aquí es que, como el principio básico de Spark RDD es la inmutabilidad. La partición o fusión creará un nuevo RDD. El RDD base seguirá existiendo con su número original de particiones. En caso de que el caso de uso exija conservar el RDD en la memoria caché, 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 :26
scala> res16.partitions.length
res17: Int = 10
scala> pairMrkt.partitions.length
res20: Int = 2
Si entiendes que te ha sido de provecho nuestro post, agradeceríamos que lo compartas con otros seniors de este modo contrubuyes a extender nuestra información.