En esta página, se explica el ajuste del rendimiento para las operaciones de JOIN en Cloud Data Fusion.
Las operaciones de JOIN pueden ser la parte más costosa de una canalización. Al igual que todo lo demás en una canalización, las operaciones se ejecutan en paralelo. El primer paso de una JOIN es redistribuir los datos para que cada registro con la misma clave JOIN se envíe al mismo ejecutor. Una vez que todos los datos se redistribuyen, se unen y el resultado continúa a través de la canalización.
Ejemplo de procesamiento en paralelo en operaciones JOIN
Por ejemplo, supongamos que realizas una operación JOIN en conjuntos de datos llamados Purchases y Items. Cada registro de compra contiene el nombre y la cantidad del artículo comprado. Cada registro de artículo contiene el nombre y el precio del artículo. Se realiza una JOIN en el nombre del artículo para calcular el precio total de cada compra. Cuando se unen los datos, se redistribuyen en el clúster de modo que los registros con el mismo ID terminen en el mismo ejecutor.
Cuando las claves de JOIN se distribuyen de manera uniforme, las operaciones de JOIN funcionan bien porque se pueden ejecutar en paralelo.
Al igual que cualquier redistribución, el sesgo de datos afecta negativamente el rendimiento. En el ejemplo anterior, los huevos se compran con mucha más frecuencia que el pollo o la leche, lo que significa que el ejecutor que une las compras de huevos realiza más trabajo que los otros ejecutores. Si notas que un JOIN está sesgado, hay dos maneras de mejorar el rendimiento.
Cómo dividir automáticamente particiones sesgadas
Con la ejecución de consultas adaptable, se controlarán automáticamente los sesgos muy grandes.
En cuanto un JOIN produce algunas particiones mucho más grandes que otras, se dividen en otras más pequeñas. Para confirmar que tienes habilitada la ejecución de consultas adaptables, consulta Ajuste automático.
Usa un JOIN en la memoria
Se puede realizar un JOIN en la memoria si un lado del JOIN es lo suficientemente pequeño como para caber en la memoria. En esta situación, el conjunto de datos pequeño se carga en la memoria y, luego, se transmite a cada ejecutor. El conjunto de datos grande no se mezcla en absoluto, lo que quita las particiones irregulares que se generan cuando se mezclan en la clave JOIN.
En el ejemplo anterior, el conjunto de datos de artículos se carga primero en la memoria del controlador de Spark. Luego, se transmite a cada ejecutor. Los ejecutores ahora pueden unir los datos sin mezclar ninguno de los conjuntos de datos de compras.
Este enfoque requiere que le proporciones suficiente memoria al controlador de Spark y a los ejecutores para que puedan almacenar el conjunto de datos de transmisión en la memoria. De forma predeterminada,
Spark reserva un poco menos del 30% de su memoria para almacenar este tipo de
datos. Cuando uses JOIN en la memoria, multiplica el tamaño del conjunto de datos por cuatro y configúralo como la memoria del ejecutor y del controlador. Por ejemplo, si el conjunto de datos de artículos fuera de 1 GB, tendríamos que establecer la memoria del ejecutor y del controlador en al menos 4 GB. No se pueden cargar conjuntos de datos de más de 8 GB en la memoria.
Distribución de claves
Cuando ambos lados de JOIN son demasiado grandes para caber en la memoria, se puede usar una técnica diferente para dividir cada clave JOIN en varias claves para aumentar el nivel de paralelismo. Esta técnica se puede aplicar a las operaciones INNER JOIN y LEFT OUTER JOIN. No se puede usar para operaciones FULL OUTER JOIN.
En este enfoque, el lado sesgado se sala con una nueva columna de números enteros con un número aleatorio de 1 a N. El lado no sesgado se expande, y cada fila existente genera N filas nuevas. Se agrega una columna nueva al lado expandido, propagada con cada número del 1 al N. Luego, se realiza un JOIN normal, excepto que la columna nueva se agrega como parte de la clave JOIN. De esta manera, todos los datos que solían ir a una sola partición ahora se distribuyen en hasta N particiones diferentes.
En el ejemplo anterior, el factor de distribución N se establece en 3. Los conjuntos de datos originales se muestran a la izquierda. Las versiones con sal y divididas del conjunto de datos se muestran en el medio. Los datos aleatorios se muestran a la derecha, con tres ejecutores diferentes que se unen a las compras de huevos, en lugar de uno.
Se logra un mayor paralelismo aumentando las distribuciones. Sin embargo, esto tiene el costo de dividir un lado del JOIN, lo que genera más datos en el clúster. Por este motivo, el beneficio disminuye a medida que aumenta la distribución. En la mayoría de las situaciones, configúralo en 20 o menos.
¿Qué sigue?
- Obtén más información sobre el procesamiento en paralelo en Cloud Data Fusion.