Optimización de las mezclas

Algunas operaciones, como join() y groupByKey(), requieren que Spark lleve a cabo una mezcla. La mezcla es el mecanismo de Spark para redistribuir los datos de forma que se agrupen de forma diferente en las particiones de RDD. Las mezclas pueden ayudarlo a corregir los cuellos de botella de rendimiento. Sin embargo, dado que la mezcla suele implicar la copia de datos entre los ejecutores de Spark, se trata de una operación compleja y costosa. Por ejemplo, las mezclas generan los siguientes costos:

La mezcla es uno de los factores más importantes en la degradación del rendimiento de su aplicación de Spark. Al almacenar los datos intermedios, puede agotar el espacio en el disco local del ejecutor, lo que provoca un error en el trabajo de Spark.

Puede evaluar el rendimiento de la mezcla en las métricas de CloudWatch y en la IU de Spark.

Métricas de CloudWatch

Si el valor de Bytes de mezcla escritos es alto en comparación con Bytes de mezcla leídos, su trabajo de Spark podría utilizar operaciones de mezcla, como join() o groupByKey().

Interfaz de usuario de Spark

En la pestaña Etapa de la IU de Spark, puede comprobar los valores de Tamaño de lectura de la mezcla / Registros. También puede verlos en la pestaña Ejecutores.

En la siguiente captura de pantalla, cada ejecutor intercambia aproximadamente 18,6 GB/4 020 000 registros con el proceso de mezcla, lo que supone un tamaño total de lectura de mezcla de unos 75 GB.

En la columna Volcado de mezcla (disco) se muestra una gran cantidad de datos que se vuelca de la memoria al disco, lo que puede provocar que el disco se llene o que se produzca un problema de rendimiento.

Si observa estos síntomas y la etapa tarda demasiado en comparación con sus objetivos de rendimiento, o si se producen los errores Out Of Memory o No space left on device, considere las siguientes soluciones.

Optimización de la unión

La operación join(), que une tablas, es la operación de mezcla más utilizada, pero suele provocar un cuello de botella de rendimiento. Como la unión es una operación costosa, le recomendamos no utilizarla a menos que sea esencial para los requisitos de su empresa. Haga las siguientes preguntas para comprobar que está haciendo un uso eficiente de su canalización de datos:

Tras confirmar que las operaciones de unión son esenciales para los requisitos de su empresa, consulte las siguientes opciones para optimizar la unión de forma que se ajuste a sus requisitos.

Uso de una inserción antes de una unión

Filtre las filas y columnas innecesarias en el DataFrame antes de llevar a cabo una unión. Esto tiene las siguientes ventajas:

# Default
df_joined  = df1.join(df2, ["product_id"])

# Use Pushdown
df1_select = df1.select("product_id","product_title","star_rating").filter(col("star_rating")>=4.0)
df2_select = df2.select("product_id","category_id")
df_joined  = df1_select.join(df2_select, ["product_id"])

Uso de una unión de DataFrame

Intente usar una API de alto nivel de Spark, como SparkSQL, DataFrame y Datasets, en lugar de la API de RDD o la unión de DynamicFrame. Puede convertir DynamicFrame en DataFrame con una llamada a un método como dyf.toDF(). Como se explica en la sección Temas clave de Apache Spark, estas operaciones de unión aprovechan internamente la optimización de consultas del optimizador Catalyst.

Uniones de mezcla y hash de transmisión y sugerencias

Spark admite dos tipos de unión: unión de mezcla y unión hash de transmisión. Una unión hash de transmisión no requiere mezclas y puede requerir menos procesamiento que una unión de mezcla. Sin embargo, solo se aplica al unir una tabla pequeña a una grande. Al unir una tabla que quepa en la memoria de un solo ejecutor de Spark, considere la posibilidad de usar una unión hash de transmisión.

En el siguiente diagrama se muestran la estructura de alto nivel y los pasos de una unión hash de transmisión y una unión de mezcla.

Los detalles de cada unión son los siguientes:

En la versión 3.0 y posteriores de AWS Glue, puede aprovechar las uniones hash de transmisión; para ello, habilite la ejecución adaptativa de consultas y otros parámetros. La ejecución adaptativa de consultas convierte una unión de ordenación/combinación en una unión hash de transmisión cuando las estadísticas de tiempo de ejecución de cualquiera de los lados de la unión son inferiores al umbral de unión hash de transmisión adaptativo.

En AWS Glue 3.0, puede activar la ejecución adaptativa de consultas mediante la definición de spark.sql.adaptive.enabled=true. La ejecución adaptativa de consultas está habilitada de forma predeterminada en AWS Glue 4.0.

Puede establecer parámetros adicionales relacionados con las mezclas y las uniones hash de transmisión:

Para obtener más información sobre los parámetros relacionados, consulte Conversión de uniones de ordenación/combinación en uniones de transmisión.

En la versión 3.0 o posteriores de AWS Glue, puede usar otras sugerencias de unión para mezclar y así ajustar el comportamiento.

-- Join Hints for shuffle sort merge join
SELECT /*+ SHUFFLE_MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGEJOIN(t2) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle hash join
SELECT /*+ SHUFFLE_HASH(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle-and-replicate nested loop join
SELECT /*+ SHUFFLE_REPLICATE_NL(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

Uso de la asignación de buckets

La unión de ordenación/combinación requiere dos fases: mezcla y ordenación y, a continuación, fusión. Estas dos fases pueden sobrecargar el ejecutor de Spark y provocar problemas de OOM y rendimiento cuando algunos ejecutores se fusionan y otros se ordenan simultáneamente. En esos casos, podría ser posible llevar a cabo una unión eficiente mediante la asignación de buckets. La asignación de buckets mezclará y ordenará previamente las entradas en las claves de unión y, a continuación, escribirá esos datos ordenados en una tabla intermedia. El costo de los pasos de mezcla y ordenación se puede reducir al unir tablas grandes definiendo las tablas intermedias ordenadas con antelación.

Las tablas en buckets son útiles para lo siguiente:

Puede crear una tabla en buckets mediante el siguiente código.

df.write.bucketBy(50, "account_id").sortBy("age").saveAsTable("bucketed_table")

Cómo volver a particionar DataFrames en las claves de unión antes de la unión

Para volver a particionar los dos DataFrames en las claves de unión antes de la unión, utilice las siguientes instrucciones.

df1_repartitioned = df1.repartition(N,"join_key")
df2_repartitioned = df2.repartition(N,"join_key")
df_joined  = df1_repartitioned.join(df2_repartitioned,"product_id")

Esto particionará dos RDD (aún separados) en la clave de unión antes de iniciar la unión. Si los dos RDD se particionan en la misma clave y con el mismo código de partición, es muy probable que los registros de RDD que desee unir estén ubicados en el mismo nodo de trabajo antes de mezclarse para la unión. Esto podría mejorar el rendimiento al reducir la actividad de la red y el sesgo de datos durante la unión.

Cómo superar el sesgo de datos

El sesgo de datos es una de las causas más habituales de los cuellos de botella para los trabajos de Spark. Se produce cuando los datos no se distribuyen uniformemente en las particiones de RDD. Esto hace que las tareas de esa partición tarden mucho más que otras, lo que retrasa el tiempo total de procesamiento de la aplicación.

Para identificar el sesgo de datos, evalúe las siguientes métricas en la IU de Spark:

Si observa un sesgo de datos, pruebe lo siguiente:

Uso de la caché

Cuando utilice DataFrames repetitivos, evite la mezcla o los cálculos adicionales mediante df.cache() o df.persist() para almacenar en caché los resultados del cálculo en la memoria de cada ejecutor de Spark y en el disco. Spark también admite la conservación de los RDD en el disco o la replicación en varios nodos (nivel de almacenamiento).

Por ejemplo, puede conservar los DataFrames agregando df.persist(). Cuando la caché ya no sea necesaria, puede utilizar unpersist para descartar los datos almacenados en caché.

df = spark.read.parquet("s3://<Bucket>/parquet/product_category=Books/")
df_high_rate = df.filter(col("star_rating")>=4.0)
df_high_rate.persist()

df_joined1 = df_high_rate.join(<Table1>, ["key"])
df_joined2 = df_high_rate.join(<Table2>, ["key"])
df_joined3 = df_high_rate.join(<Table3>, ["key"])
... 
df_high_rate.unpersist()

Eliminación de las acciones innecesarias de Spark

Evita ejecutar acciones innecesarias como count, show o collect. Como se explica en la sección Temas clave de Apache Spark, Spark es lento. Cada RDD transformado puede volver a calcularse cada vez que ejecute una acción en él. Cuando utiliza muchas acciones de Spark, se llama a varios accesos a orígenes, cálculos de tareas y ejecuciones de mezcla para cada acción.

Si no necesita collect() ni otras acciones en su entorno comercial, considere la posibilidad de eliminarlas.