Monter en charge les clusters Ray sur Databricks

Découvrez comment optimiser la taille de votre cluster Ray pour des performances optimales, notamment l'autoscaling, la configuration du nœud principal, les clusters hétérogènes et l'allocation des ressources.

Créez un cluster Ray en mode de mise à l'échelle automatique

Dans Ray 2.8.0 et versions ultérieures, les clusters Ray start sur Databricks prennent en charge l'intégration avec l'autoscaling Databricks. Cette intégration d'évolutivité automatique Trigger l'évolutivité automatique des clusters Databricks en interne au sein de l'environnement Databricks.

Pour activer la mise à l'échelle automatique, exécutez la commande suivante :

Pour la version de Ray inférieure à 2.10 :

Python

from ray.util.spark import setup_ray_cluster

setup_ray_cluster(
  num_worker_nodes=8,
  autoscale=True,
)

Pour la version de Ray 2,10 et suivantes :

Python

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="/Volumes/<catalog>/<schema>/<volume>/path/to/ray_collected_logs"
)

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

L'API ray.util.spark.setup_ray_cluster crée un cluster Ray sur Apache Spark. En interne, il crée un Apache Spark Job en arrière-plan. Chaque tâche Apache Spark du job crée un nœud Ray worker, et le nœud principal Ray est créé sur le driver. Les arguments min_worker_nodes et max_worker_nodes représentent la plage de nœuds Ray worker à créer et à utiliser pour les charges de travail Ray. Si l'argument min_worker_nodes n'est pas défini, un cluster Ray de taille fixe sera start avec max_worker_nodes workers disponibles. Pour spécifier le nombre de cœurs CPU ou GPU attribués à chaque nœud Ray Worker, définissez l'argument num_cpus_worker_node (valeur par default : 1) ou num_gpus_worker_node (valeur par default : 0).

Pour les versions de Ray antérieures à 2.10, si le dimensionnement automatique est activé, num_worker_nodes indique le nombre maximal de nœuds worker Ray. Le nombre minimum default de nœuds Ray Worker est zéro. Ce paramètre default signifie que lorsque le cluster Ray est inactif, il se réduit à zéro nœud Ray Worker. Cela pourrait ne pas être idéal pour une réactivité rapide dans tous les scénarios, mais cela peut réduire considérablement les coûts lorsque c'est activé.

En mode de dimensionnement automatique, num_worker_nodes ne peut pas être défini sur ray.util.spark.MAX_NUM_WORKER_NODES.

Les arguments suivants configurent la vitesse de mise à l'échelle et de réduction de la mise à l'échelle :

• autoscale_upscaling_speed représente le nombre de nœuds autorisés à être en attente en tant que multiple du nombre actuel de nœuds. Plus la valeur est élevée, plus la mise à l'échelle est agressive. Par exemple, si cette valeur est définie sur 1,0, le cluster peut augmenter en taille d'au plus 100 % à tout moment.
• autoscale_idle_timeout_minutes représente le nombre de minutes qui doivent s'écouler avant que l'autoscaler ne supprime un nœud Worker inactif. Plus la valeur est petite, plus la réduction d'échelle est agressive.

Avec Ray 2.9.0 et versions ultérieures, vous pouvez également définir autoscale_min_worker_nodes pour empêcher le cluster Ray de se réduire à zéro worker lorsque le cluster Ray est inactif, ce qui entraînerait l'arrêt du cluster.

Configurer les ressources utilisées par le nœud principal Ray

Par default, pour la configuration Ray sur Spark, Databricks limite les ressources allouées au nœud principal Ray à :

• 0 cœurs de CPU
• 0 GPU
• 128 Mo de mémoire de tas
• 128 Mo de mémoire de stockage d'objets

C'est parce que le nœud principal Ray n'est généralement utilisé que pour la coordination globale, et non pour l'exécution des tâches Ray. Les ressources du nœud driver Apache Spark sont partagées avec plusieurs utilisateurs, donc le paramètre default économise les ressources du côté du driver Apache Spark. Avec Ray 2.8.0 et versions ultérieures, vous pouvez configurer les ressources utilisées par le nœud principal Ray. Utilisez les arguments suivants dans l'API setup_ray_cluster :

• num_cpus_head_node: configuration des cœurs de processeur utilisés par le nœud principal Ray
• num_gpus_head_node: Paramétrage du GPU utilisé par le nœud principal Ray
• object_store_memory_head_node: définition de la taille de la mémoire de stockage d'objets par le nœud principal Ray

Prise en charge des clusters hétérogènes

Vous pouvez créer un cluster Ray sur Spark pour des exécutions d'entraînement plus efficaces et plus rentables, et définir différentes configurations entre le nœud principal Ray et les nœuds Worker Ray. Cependant, tous les nœuds Worker Ray doivent avoir la même configuration. Les clusters Databricks ne prennent pas entièrement en charge les clusters hétérogènes, mais vous pouvez créer un cluster Databricks avec différents types d'instances de Driver et de Worker en définissant une règle de cluster. Par exemple :

{
  "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"
  }
}

Réglez la configuration du cluster Ray

La configuration recommandée pour chaque nœud worker Ray est la suivante : Un minimum de 4 cœurs de CPU par nœud worker Ray. Mémoire de tas minimale de 10 Go pour chaque nœud worker Ray.

Ainsi, lors de l'appel de ray.util.spark.setup_ray_cluster, Databricks recommande de définir num_cpus_per_node à une valeur supérieure ou égale à 4.

Consultez la section suivante pour plus de détails sur le réglage de la mémoire de segment pour chaque nœud Worker Ray.

Allocation de mémoire pour les nœuds Worker Ray

Chaque nœud worker Ray utilise deux types de mémoire : la mémoire de tas et la mémoire du magasin d'objets.

La taille de mémoire allouée pour chaque type est déterminée comme décrit ci-dessous.

La mémoire totale allouée à chaque nœud worker Ray est : 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 est le nombre maximal de nœuds worker Ray pouvant être lancés sur le nœud worker Apache Spark. Ceci est déterminé par l'argument num_cpus_per_node ou num_gpus_per_node.

Si vous ne définissez pas l'argument object_store_memory_per_node, alors la taille de la mémoire de tas et la taille de la mémoire du magasin d'objets allouées à chaque nœud Worker Ray sont : RAY_WORKER_NODE_HEAP_MEMORY = RAY_WORKER_NODE_TOTAL_MEMORY * 0.7 OBJECT_STORE_MEMORY_PER_NODE = RAY_WORKER_NODE_TOTAL_MEMORY * 0.3

Si vous définissez l'argument object_store_memory_per_node: RAY_WORKER_NODE_HEAP_MEMORY = RAY_WORKER_NODE_TOTAL_MEMORY - argument_object_store_memory_per_node

En outre, la taille de la mémoire du magasin d’objets par nœud Worker Ray est limitée par la mémoire partagée du système d’exploitation. La valeur maximale est : 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 est la taille de disque /dev/shm configurée pour le nœud worker Apache Spark.

Mise à l'échelle des bonnes pratiques

Définir le nombre de CPU et de GPU pour chaque Ray Worker

Nous vous recommandons de définir l'argument num_cpus_worker_node sur le nombre de cœurs de CPU par nœud worker Apache Spark. De même, la définition de num_gpus_worker_node sur le nombre de GPU par nœud worker Apache Spark est optimale. Avec cette configuration, chaque nœud worker Apache Spark lance un nœud worker Ray qui utilisera pleinement les ressources de chaque nœud worker Apache Spark.

Définissez la variable d'environnement RAY_memory_monitor_refresh_ms à 0 dans la configuration du cluster Databricks lorsque vous démarrez votre cluster Apache Spark.

Configuration des ressources mémoire pour les charges de travail hybrides Apache Spark et Ray

Si vous exécutez des charges de travail hybrides Spark et Ray dans un cluster Databricks, Databricks vous recommande de réduire la mémoire de l'exécuteur Spark à une petite valeur. Par exemple, définissez spark.executor.memory 4g dans la configuration du cluster Databricks.

L'exécuteur Apache Spark est un processus Java qui Trigger la GC de manière paresseuse, et le cache de dataset Apache Spark utilise beaucoup de mémoire de l'exécuteur Apache Spark. Cela réduit la mémoire disponible que Ray peut utiliser. Pour éviter d'éventuelles erreurs de mémoire insuffisante, réduisez la configuration spark.executor.memory.

Configuration des Ressources de calcul pour les charges de travail hybrides Apache Spark et Ray

Si vous exécutez des charges de travail hybrides Spark et Ray dans un cluster Databricks, nous vous recommandons de rendre les nœuds de cluster ou les nœuds worker Ray auto-évolutifs. Par exemple :

Si vous disposez d'un nombre fixe de nœuds Worker disponibles pour start un cluster Databricks, nous vous recommandons d'activer l'autoscaling Ray-on-Spark. Lorsqu'aucune charge de travail Ray n'est en cours d'exécution, le cluster Ray est mis à l'échelle, ce qui permet de libérer des ressources pour qu'elles soient utilisées par les tâches Apache Spark. Lorsque les tâches Apache Spark sont terminées et que Ray est à nouveau utilisé, le cluster Ray-on-Spark sera à nouveau mis à l'échelle pour répondre à la demande.

De plus, vous pouvez rendre les clusters Databricks et Ray-on-spark évolutifs automatiquement. Par exemple, si vous configurez les nœuds auto-évolutifs du cluster Databricks à un maximum de 10 nœuds, configurez les nœuds worker Ray-on-Spark à un maximum de quatre nœuds, et configurez chaque nœud worker Ray pour utiliser pleinement les ressources de chaque worker Apache Spark, les charges de travail Ray peuvent utiliser au plus quatre nœuds de ressources sur une telle configuration de cluster. En comparaison, les jobs Apache Spark peuvent allouer au maximum six nœuds de ressources.