Aller au contenu principal
Dernière mise à jour le

Créer et se connecter aux clusters Ray sur Databricks

Apprenez à créer, configurer et exécuter des clusters de compute Ray sur Databricks

Exigences​

Pour créer un cluster Ray, vous devez avoir accès à une ressource de compute multifonction Databricks avec les paramètres suivants :

Installer Ray​

À partir de Databricks Runtime ML 15.0, Ray est préinstallé sur les clusters Databricks.

Pour les runtimes publiés avant la version 15.0, utilisez pip pour installer Ray sur votre cluster :

%pip install ray[default]>=2.3.0

Créer un cluster Ray spécifique à l'utilisateur dans un cluster Databricks​

Pour créer un cluster Ray, utilisez ray.util.spark.setup_ray_cluster API.

remarque

Lorsque vous créez un cluster Ray dans un Notebook, il est uniquement disponible pour l'utilisateur de Notebook actuel. Le cluster Ray est automatiquement arrêté après que le Notebook est détaché du cluster ou après 30 minutes d'inactivité (aucune tâche n'a été soumise à Ray). Si vous souhaitez créer un cluster Ray qui est partagé avec tous les utilisateurs et qui n'est pas soumis à un notebook en cours d'exécution, utilisez plutôt l'API ray.util.spark.setup_global_ray_cluster.

Cluster Ray de taille fixe​

Dans tout notebook Databricks qui est associé à un cluster Databricks, vous pouvez exécuter la commande suivante pour start un cluster Ray de taille fixe :

Python
import ray
from ray.util.spark import setup_ray_cluster, shutdown_ray_cluster

setup_ray_cluster(
  max_worker_nodes=1,
  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)

Cluster Ray à mise à l'échelle automatique​

Pour savoir comment start un cluster Ray à mise à l'échelle automatique, consultez Monter en charge les clusters Ray sur Databricks.

Démarrage d'un cluster Ray en mode global​

Avec Ray 2.9.0 et versions ultérieures, vous pouvez créer un cluster Ray en mode global sur un cluster Databricks. Un cluster Ray en mode global permet à tous les utilisateurs attachés à la ressource de compute Databricks d'utiliser également le cluster Ray. Ce mode d'exécution d'un cluster Ray n'a pas la fonctionnalité de délai d'attente actif qu'une ressource de compute dédiée possède lors de l'exécution d'une instance de cluster Ray mono-utilisateur.

Pour start un cluster Ray global auquel plusieurs utilisateurs peuvent se connecter et exécuter des tâches Ray, commencez par créer un Job de Notebook Databricks et associez-le à un cluster Databricks en mode partagé, puis exécutez la commande suivante :

Python
from ray.util.spark import setup_global_ray_cluster

setup_global_ray_cluster(
  max_worker_nodes=2,
  ...
  # other arguments are the same as with the `setup_global_ray` API.
)

Il s'agit d'un appel bloquant qui restera actif jusqu'à ce que vous interrompiez l'appel en cliquant sur le bouton « Interrupt » de la cellule de commande du Notebook, en détachant le Notebook du cluster Databricks ou en mettant fin au cluster Databricks. Autrement, le cluster Ray en mode global continuera à fonctionner et sera disponible pour la soumission de tâches par les utilisateurs autorisés. Pour plus d'informations sur les clusters en mode global, consultez la documentation de l'API Ray.

Les clusters en mode global ont les propriétés suivantes :

  • Dans un cluster Databricks, vous ne pouvez créer qu'un seul cluster Ray de mode global actif à la fois.
  • Dans un cluster Databricks, le cluster Ray en mode global actif peut être utilisé par tous les utilisateurs dans tout notebook Databricks attaché. Vous pouvez exécuter ray.init() pour vous connecter au cluster Ray global actif. Étant donné que plusieurs utilisateurs peuvent accéder à ce cluster Ray, la contention des ressources peut poser problème.
  • Le cluster Ray en mode global est actif jusqu'à ce que l'appel setup_ray_cluster soit interrompu. Il ne dispose pas d'un délai d'arrêt automatique comme les clusters Ray mono-utilisateur.

Créer un cluster GPU Ray​

Pour les clusters GPU, ces ressources peuvent être ajoutées au cluster Ray de la manière suivante :

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=8,
  num_gpus_per_node=1,
  num_cpus_head_node=8,
  num_gpus_head_node=1,
  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)

Se connecter à un cluster Ray distant à l'aide du client Ray​

Dans Ray version 2.3.0 et versions ultérieures, vous pouvez créer un cluster Ray à l'aide de l'API setup_ray_cluster, et dans le même notebook, vous pouvez appeler ray.init(). API pour se connecter à ce cluster Ray. Pour obtenir la chaîne de connexion à distance, utilisez les éléments suivants :

Python
from ray.util.spark import setup_ray_cluster

_, remote_conn_str = setup_ray_cluster(num_worker_nodes=2, ...)

Ensuite, vous pouvez connecter le cluster distant à l’aide de la chaîne de connexion distante ci-dessus :

Python
import ray
ray.init(remote_conn_str)

Le client Ray ne prend pas en charge l'API Ray dataset définie dans le module ray.data. Comme solution de contournement, vous pouvez encapsuler votre code qui appelle l'API Ray dataset dans une tâche Ray distante, comme indiqué dans le code suivant :

Python
import ray
import pandas as pd

# Note: This must be run in the same VPC/network as the Spark cluster
# so it can reach this address
ray.init("ray://<ray_head_node_ip>:10001")

@ray.remote
def ray_data_task():
    p1 = pd.DataFrame({'a': [3,4] * 10000, 'b': [5,6] * 10000})
    ds = ray.data.from_pandas(p1)
    return ds.repartition(4).to_pandas()

ray.get(ray_data_task.remote())
## Connecting the Ray Cluster to the Ray Job CLI

For many developers moving from self-managed Ray solutions to a <Databricks> solution, there is often existing infrastructure tooling built based on the Ray CLI tools. While <Databricks> currently does not support Ray Cluster CLI integration, the Ray Job CLI can be connected through the driver proxy to the Ray cluster running on <Databricks>. For example:

``` shell
ray job submit  --headers '{"cookie" : "DATAPLANE_DOMAIN_SESSIONID=<REDACTED>"}' --address 'https://<DATABRICKS WORKSPACE URL>/driver-proxy/o/<etc>' --working-dir='.' -- python run_task.py

Les valeurs à configurer sont l'URL du Workspace Databricks, commençant par https://, puis les valeurs trouvées après le /driver-proxy/o/ se trouvent dans l'URL du proxy du tableau de bord Ray affichée après le start du cluster Ray.

Le Ray Job CLI est utilisé pour soumettre des Jobs à un cluster Ray depuis des systèmes externes, mais il n'est pas requis pour soumettre des Jobs sur des clusters Ray sur Databricks. Il est recommandé que le Job soit déployé à l'aide de Lakeflow Jobs, qu'un cluster Ray par application soit créé et que les outils Databricks existants, tels que Databricks Asset Bundles ou Workflow Triggers, soient utilisés pour Trigger le Job.

Définir un emplacement de sortie de Logs​

Vous pouvez définir l'argument collect_log_to_path pour spécifier le chemin de destination où vous souhaitez collecter les Logs du cluster Ray. La collecte des logs s'exécute après l'arrêt du cluster Ray.

Databricks recommande de définir un chemin commençant par /dbfs/ ou un chemin de volume Unity Catalog pour conserver les logs même si vous arrêtez le cluster Apache Spark. Sinon, vos logs ne sont pas récupérables, car le stockage local du cluster est supprimé lorsque le cluster est arrêté.

Après avoir créé un cluster Ray, vous pouvez exécuter n'importe quel code d'application Ray directement dans votre Notebook. Cliquez sur Ouvrir le tableau de bord du cluster Ray dans un nouvel tab pour afficher le tableau de bord Ray du cluster.

Activer les traces de pile et les graphes de flammes sur la page Ray Dashboard Actors​

Sur la page Acteurs du tableau de bord Ray, vous pouvez afficher les traces de la pile et les graphes de flammes pour les acteurs Ray actifs. Pour consulter ces informations, utilisez la commande suivante pour installer py-spy avant de start le cluster Ray :

Python
%pip install py-spy

Créer et configurer les meilleures pratiques​

Cette section couvre les meilleures pratiques pour la création et la configuration de clusters Ray.

Charges de travail non-GPU​

Le cluster Ray s'exécute sur un cluster Databricks Spark. Un scénario typique est d'utiliser un Job Spark et une UDF Spark pour effectuer des tâches simples de prétraitement des données qui ne nécessitent pas de Ressources GPU. Ensuite, utilisez Ray pour exécuter des tâches de Machine Learning compliquées qui bénéficient des GPU. Dans ce cas, Databricks recommande de définir le parameter de configuration au niveau du cluster Apache Spark spark.task.resource.gpu.amount sur 0 afin que toutes les Transformations de DataFrame Apache Spark et les exécutions d'UDF Apache Spark n'utilisent pas de Ressources GPU.

Les avantages de cette configuration sont les suivants :

  • Cela augmente le parallélisme des Jobs Apache Spark, car le type d'instance GPU dispose généralement de beaucoup plus de cœurs de CPU que de périphériques GPU.
  • Si le cluster Apache Spark est partagé avec plusieurs utilisateurs, cette configuration empêche les jobs Apache Spark d’entrer en concurrence avec les charges de travail Ray exécutées simultanément pour les ressources GPU.

Désactiver l'intégration MLflow du formateur transformers si vous l'utilisez dans des tâches Ray​

L'intégration MLflow de l'entraîneur transformers est activée par default à partir de la bibliothèque transformers. Si vous utilisez Ray train pour affiner un modèle transformers, les tâches Ray échoueront en raison d'un problème d'informations d'identification. Cependant, ce problème ne s'applique pas si vous utilisez directement MLflow pour la formation. Pour éviter ce problème, vous pouvez définir la variable d'environnement DISABLE_MLFLOW_INTEGRATION sur 'TRUE' à partir de la configuration du cluster Databricks lors du démarrage de votre cluster Apache Spark.

Corriger l'erreur de sérialisation de la fonction distante Ray​

Pour exécuter les tâches Ray, Ray *pickles* la fonction de tâche. Si vous constatez l'échec du pickling, vous devez diagnostiquer quelle partie de votre code est à l'origine de l'échec. Les causes courantes des erreurs de pickling sont la gestion des références externes, des closures et des références aux objets stateful. L'une des erreurs les plus faciles à vérifier et à corriger rapidement peut être résolue en déplaçant les instructions d'importation dans la déclaration de fonction de tâche.

Par exemple, datasets.load_dataset est une fonction largement utilisée qui est patchée côté driver du Databricks Runtime, ce qui rend la référence impossible à désérialiser. Pour y remédier, vous pouvez simplement écrire la fonction de tâche comme suit :

Python
def ray_task_func():
  from datasets import load_dataset  # import the function inside task function
  ...

Désactiver le moniteur de mémoire Ray si la tâche Ray est arrêtée de manière inattendue avec une erreur de mémoire (OOM)​

Dans Ray 2.9.3, le moniteur de mémoire Ray présente plusieurs problèmes connus qui peuvent entraîner l'arrêt involontaire des tâches Ray sans raison. Pour résoudre le problème, vous pouvez désactiver le moniteur de mémoire Ray en définissant la variable d’environnement RAY_memory_monitor_refresh_ms sur 0 dans la configuration du cluster Databricks lors du démarrage de votre cluster Apache Spark.

Application de fonctions de transformation aux batch de données​

Lors du traitement des données en batch, il est recommandé d'utiliser l'API Ray Data avec la fonction map_batches. Cette approche peut être plus efficace et évolutive, en particulier pour les grands datasets ou les calculs complexes qui bénéficient du traitement par batch. Tout DataFrame Spark peut être converti en Dataset Ray à l'aide de l'API ray.data.from_spark. La sortie traitée issue de l'appel de cette API de transformation peut être écrite dans des tables UC Databricks à l'aide de l'API ray.data.write_databricks_table.

Utilisation de MLflow dans Ray tuner, Ray train ou les tâches Ray personnalisées​

L'intégration de Databricks MLflow et Ray nécessite Ray 2,41 et versions ultérieures.

Pour utiliser MLflow avec Ray Tune, Ray Train ou des tâches Ray personnalisées, définissez les variables d'environnement suivantes : DATABRICKS_HOST et DATABRICKS_TOKEN, ou définissez les variables d'environnement DATABRICKS_HOST, DATABRICKS_CLIENT_ID et DATABRICKS_CLIENT_SECRET avant d'appeler ray.util.spark.setup_ray_cluster. Le code suivant montre comment définir ces variables.

Python
import os
from ray.util.spark import setup_ray_cluster

os.environ["DATABRICKS_HOST"] = "https://....databricks.com"
os.environ["DATABRICKS_TOKEN"] = "<your PAT token"

setup_ray_cluster(num_cpus_worker_node=2, num_gpus_worker_node=0, max_worker_nodes=1, min_worker_nodes=1)

Utilisez les bibliothèques Python délimitées par un Notebook ou les bibliothèques Python de cluster dans les tâches Ray​

L'utilisation de bibliothèques Python au niveau du notebook ou de bibliothèques Python de cluster dans les tâches Ray distantes nécessite Ray 2.12 ou une version ultérieure.

Les versions de Ray 2.11 et antérieures présentent un problème connu selon lequel les tâches Ray ne peuvent pas utiliser les bibliothèques Python délimitées par un Notebook ou les bibliothèques Python de clusters. Pour les versions de Ray 2.11 et antérieures, des dépendances supplémentaires doivent pré-installer des dépendances dans la session active à l’aide de la commande magique %pip avant de démarrer le clusters Ray.

Ressources supplémentaires​

Sur cette page