Escalado de clústeres de Ray en Azure Databricks

Aprenda a ajustar el tamaño del clúster de Ray para obtener un rendimiento óptimo, incluido el escalado automático, la configuración del nodo principal, los clústeres heterogéneos y la asignación de recursos.

Creación de un clúster de Ray en modo de escalado automático

En Ray 2.8.0 y versiones posteriores, los clústeres de Ray que se iniciaron en Azure Databricks admiten la integración con el escalado automático de Azure Databricks. Esta integración de escalado automático desencadena el escalado automático del clúster de Azure Databricks internamente dentro del entorno de Databricks.

Para habilitar el escalado automático, ejecute el siguiente comando:

Para la versión de Ray inferior a la 2.10:

from ray.util.spark import setup_ray_cluster

setup_ray_cluster(
  num_worker_nodes=8,
  autoscale=True,
)

Para la versión de Ray 2.10 y versiones posteriores:

from ray.util.spark import setup_ray_cluster, shutdown_ray_cluster

setup_ray_cluster(
  min_worker_nodes=2,
  max_worker_nodes=4,
  num_cpus_per_node=4,
  collect_log_to_path="/dbfs/path/to/ray_collected_logs"
)

# Pass any custom Ray configuration with ray.init
ray.init(ignore_reinit_error=True)

La API ray.util.spark.setup_ray_cluster crea un clúster de Ray en Apache Spark. Internamente, crea un trabajo de Apache Spark en segundo plano. Cada tarea de Apache Spark del trabajo crea un nodo de trabajo de Ray y el nodo principal de Ray se crea en el controlador. Los argumentos min_worker_nodes y max_worker_nodes representan el intervalo de nodos de trabajo de Ray para crear y usar para cargas de trabajo de Ray. Si el argumento min_worker_nodes se deja sin definir, se iniciará un clúster de Ray de tamaño fijo con max_worker_nodes el número de trabajos disponibles. Para especificar el número de núcleos de CPU o GPU asignados a cada nodo de trabajo de Ray, establezca el argumento num_cpus_worker_node (valor predeterminado: 1) o num_gpus_worker_node (valor predeterminado: 0).

Si el escalado automático está habilitado en las versiones de Ray inferiores a la 2.10, num_worker_nodes indica el número máximo de nodos de trabajo de Ray. El número mínimo predeterminado de nodos de trabajo de Ray es cero. Esta configuración predeterminada significa que cuando el clúster de Ray está inactivo, se reduce verticalmente a cero nodos de trabajo de Ray. Esto puede no ser ideal para una capacidad de respuesta rápida en todos los escenarios, pero puede reducir significativamente los costos cuando está habilitado.

En el modo de escalado automático, num_worker_nodes no se puede establecer en ray.util.spark.MAX_NUM_WORKER_NODES.

Los siguientes argumentos configuran la velocidad de escalado y reducción vertical:

• autoscale_upscaling_speed representa el número de nodos que pueden estar pendientes como un múltiplo del número actual de nodos. Cuanto mayor sea el valor, más agresivo será el escalado vertical. Por ejemplo, si se establece en 1,0, el clúster puede crecer en tamaño como máximo un 100 % en cualquier momento.
• autoscale_idle_timeout_minutes representa el número de minutos que deben pasar antes de que el escalador automático quite un nodo de trabajo inactivo. Cuanto menor sea el valor, más agresivo será la reducción.

Con Ray 2.9.0 y versiones posteriores, también puede establecer autoscale_min_worker_nodes para evitar que el clúster de Ray se reduzca verticalmente a cero trabajos cuando esté inactivo, lo que haría que el clúster finalice.

Configuración de recursos usados por el nodo principal de Ray

De forma predeterminada, para la configuración de Ray en Spark, Azure Databricks restringe los recursos asignados al nodo principal de Ray a:

• 0 núcleos de CPU
• 0 GPU
• Memoria en montón de 128 MB
• Memoria de almacenamiento de objetos de 128 MB

Esto se debe a que el nodo principal de Ray normalmente solo se usa para la coordinación global, no para ejecutar tareas de Ray. Los recursos del nodo del controlador de Apache Spark se comparten con varios usuarios, por lo que la configuración predeterminada guarda los recursos en el lado del controlador de Apache Spark. Con Ray 2.8.0 y versiones posteriores, puede configurar los recursos utilizados por el nodo principal de Ray. Use los argumentos siguientes en la API de setup_ray_cluster:

• num_cpus_head_node: establecer núcleos de CPU usados por el nodo principal de Ray
• num_gpus_head_node: configuración de GPU usada por el nodo principal de Ray
• object_store_memory_head_node: establecer el tamaño de memoria del almacén de objetos por el nodo principal de Ray

Compatibilidad con clústeres heterogéneos

Para ejecuciones de entrenamiento más eficaces y rentables, puede crear un clúster de Ray en Spark y establecer configuraciones diferentes entre el nodo principal de Ray y sus nodos de trabajo. Sin embargo, todos los nodos de trabajo de Ray deben tener la misma configuración. Los clústeres de Azure Databricks no admiten completamente clústeres heterogéneos, pero puede crear un clúster de Azure Databricks con diferentes tipos de instancia de controlador y de trabajo estableciendo una directiva de clúster. Por ejemplo:

{
  "node_type_id": {
    "type": "fixed",
    "value": "i3.xlarge"
  },
  "driver_node_type_id": {
    "type": "fixed",
    "value": "g4dn.xlarge"
  },
  "spark_version": {
    "type": "fixed",
    "value": "13.x-snapshot-gpu-ml-scala2.12"
  }
}

Ajuste de la configuración del clúster de Ray

La configuración recomendada para cada nodo de trabajo de Ray es la siguiente: Mínimo 4 núcleos de CPU por nodo de trabajo de Ray. Memoria de montón mínima de 10 GB para cada nodo de trabajo de Ray.

Por lo tanto, al llamar a ray.util.spark.setup_ray_cluster, Azure Databricks recomienda establecer num_cpus_per_node en un valor mayor o igual que 4.

Consulte la sección siguiente para obtener más información sobre cómo optimizar la memoria del montón para cada nodo de trabajo de Ray.

Asignación de memoria para nodos de trabajo de Ray

Cada nodo de trabajo de Ray usa dos tipos de memoria: memoria del montón y memoria de almacenamiento de objetos.

El tamaño de memoria asignado para cada tipo se determina como se describe a continuación.

La memoria total asignada a cada nodo de trabajo de Ray es: RAY_WORKER_NODE_TOTAL_MEMORY = (SPARK_WORKER_NODE_PHYSICAL_MEMORY / MAX_NUMBER_OF_LOCAL_RAY_WORKER_NODES * 0.8)

MAX_NUMBER_OF_LOCAL_RAY_WORKER_NODES es el número máximo de nodos de trabajo de Ray que se pueden iniciar en el nodo de trabajo de Apache Spark. Esto viene determinado por el argumento num_cpus_per_node o num_gpus_per_node.

Si no establece el argumento object_store_memory_per_node, el tamaño de la memoria del montón y el tamaño de la memoria del almacén de objetos asignados a cada nodo de trabajo de Ray son: RAY_WORKER_NODE_HEAP_MEMORY = RAY_WORKER_NODE_TOTAL_MEMORY * 0.7OBJECT_STORE_MEMORY_PER_NODE = RAY_WORKER_NODE_TOTAL_MEMORY * 0.3

Si establece el argumento object_store_memory_per_node: RAY_WORKER_NODE_HEAP_MEMORY = RAY_WORKER_NODE_TOTAL_MEMORY - argument_object_store_memory_per_node

Además, el tamaño de memoria del almacén de objetos por nodo de trabajo de Ray está limitado por la memoria compartida del sistema operativo. El valor máximo es: OBJECT_STORE_MEMORY_PER_NODE_CAP = (SPARK_WORKER_NODE_OS_SHARED_MEMORY / MAX_NUMBER_OF_LOCAL_RAY_WORKER_NODES * 0.8)

SPARK_WORKER_NODE_OS_SHARED_MEMORY es el tamaño del disco /dev/shm configurado para el nodo de trabajo de Apache Spark.

Procedimientos recomendados para el escalado

Establecer el número de CPU y GPU para cada nodo de trabajo de Ray

Se recomienda establecer el argumento num_cpus_worker_node en el número de núcleos de CPU por nodo de trabajo de Apache Spark. De forma similar, establecer num_gpus_worker_node en el número de GPU por nodo de trabajo de Apache Spark es óptimo. Con esta configuración, cada nodo de trabajo de Apache Spark inicia un nodo de trabajo de Ray que usará completamente los recursos de cada nodo de trabajo de Apache Spark.

Establezca la variable de entorno RAY_memory_monitor_refresh_ms en 0 dentro de la configuración del clúster de Azure Databricks al iniciar el clúster de Apache Spark.

Configuración de recursos de memoria para cargas de trabajo híbridas de Apache Spark y Ray

Si ejecuta cargas de trabajo híbridas de Spark y Ray en un clúster de Azure Databricks, Azure Databricks recomienda reducir la memoria del ejecutor de Spark a un valor pequeño. Por ejemplo, establezca spark.executor.memory 4g en la configuración del clúster de Azure Databricks.

El ejecutor de Apache Spark es un proceso de Java que desencadena GC diferido y la memoria caché del conjunto de datos de Apache Spark usa una gran cantidad de memoria del ejecutor de Apache Spark. Esto reduce la memoria disponible que Ray puede usar. Para evitar posibles errores de memoria insuficiente, reduzca la configuración spark.executor.memory.

Configuración de recursos de cálculo para cargas de trabajo híbridas de Apache Spark y Ray

Si ejecuta cargas de trabajo híbridas de Spark y Ray en un clúster de Azure Databricks, se recomienda que los nodos de clúster o los nodos de trabajo de Ray se escalen automáticamente. Por ejemplo:

Si tiene un número fijo de nodos de trabajo disponibles para iniciar un clúster de Azure Databricks, se recomienda habilitar el escalado automático de Ray en Spark. Cuando no se ejecutan cargas de trabajo de Ray, el clúster de Ray se reduce verticalmente, lo que permite liberar recursos para que los usen las tareas de Apache Spark. Cuando finalicen las tareas de Apache Spark y se vuelva a usar Ray, el clúster de Ray en Spark se volverá a escalar verticalmente para satisfacer la demanda.

Además, puede hacer que los clústeres de Azure Databricks y de Ray en Spark se puedan escalar automáticamente. Por ejemplo, si configura los nodos escalables automáticamente del clúster de Azure Databricks en un máximo de 10 nodos, configure los nodos de trabajo de Ray en Spark en un máximo de cuatro nodos y establezca un nodo de trabajo de Ray para que funcione por trabajo de Apache Spark. Las cargas de trabajo de Ray pueden usar como máximo cuatro recursos de nodos en dicha configuración de clúster. En comparación, los trabajos de Apache Spark pueden asignar como máximo seis nodos de recursos.