Utiliser Ray sur Azure Databricks

Article
04/22/2024

Avec Ray 2.3.0 et versions ultérieures, vous pouvez créer des clusters Ray et exécuter des applications Ray sur des clusters Apache Spark avec Azure Databricks. Pour plus d’informations sur la prise en main du Machine Learning sur Ray, notamment des didacticiels et des exemples, consultez la documentation Ray. Pour plus d'informations sur l'intégration de Ray et Apache Spark, consultez la documentation Ray on Spark API.

Spécifications

Databricks Runtime 12.2 LTS ML et versions ultérieures.
Le mode d’accès au cluster Databricks Runtime doit être « Attribué » ou « Aucun isolement partagé ».

Installer Ray

Utilisez la commande suivante pour installer Ray. L’extension [default] est requise par le composant de tableau de bord Ray.

%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 l’API ray.util.spark.setup_ray_cluster .

Dans n’importe quel notebook Databricks attaché à un cluster Databricks, vous pouvez exécuter la commande suivante :

from ray.util.spark import setup_ray_cluster, shutdown_ray_cluster

setup_ray_cluster(
  num_worker_nodes=2,
  num_cpus_worker_node=4,
  collect_log_to_path="/dbfs/path/to/ray_collected_logs"
)

L’API ray.util.spark.setup_ray_cluster crée un cluster Ray sur Spark. En interne, il crée un job Spark en arrière-plan. Chaque tâche Spark du Chaque tâche Spark dans la tâche crée un nœud Worker Ray, et le nœud principal Ray est créé sur le pilote.crée un nœud De travail Ray et le nœud principal Ray est créé sur le pilote. L’argument num_worker_nodes représente le nombre de nœuds Worker Ray à créer. Pour spécifier le nombre de cœurs UC ou GPU attribués à chaque nœud Worker Ray, définissez l’argument num_cpus_worker_node (valeur par défaut : 1) ou num_gpus_worker_node (valeur par défaut : 0).

Une fois qu'un cluster Ray est créé, vous pouvez exécuter directement dans votre notebook tout code d'application Ray. Cliquez sur Ouvrir le tableau de bord du cluster Ray dans un nouvel onglet pour afficher le tableau de bord Ray du cluster.

Conseil

Si vous utilisez un cluster utilisateur unique Azure Databricks, vous pouvez définir num_worker_nodes sur ray.util.spark.MAX_NUM_WORKER_NODES afin d’utiliser toutes les ressources disponibles pour votre cluster Ray.

setup_ray_cluster(
  # ...
  num_worker_nodes=ray.util.spark.MAX_NUM_WORKER_NODES,
)

Définissez l’argument collect_log_to_path pour spécifier le chemin de destination où vous souhaitez collecter les journaux du cluster Ray. La collecte des journaux s'exécute après l'arrêt du cluster Ray. Databricks recommande de définir un chemin commençant par /dbfs/ afin que les journaux soient préservés même si vous mettez fin au cluster Spark. Sinon, vos journaux ne sont pas récupérables, car le stockage local sur le cluster est supprimé lorsque le cluster est arrêté.

Remarque

« Pour que votre application Ray utilise automatiquement le cluster Ray qui a été créé, appelez ray.util.spark.setup_ray_cluster pour définir la variable d’environnement RAY_ADDRESS sur l’adresse du cluster Ray. » Vous pouvez spécifier une adresse de cluster alternative en utilisant l'argument address de l'API ray.init.

Exécuter une application Ray

Une fois que le cluster Ray a été créé, vous pouvez exécuter n'importe quel code d'application Ray dans un notebook Azure Databricks.

Important

Databricks recommande d'installer toutes les bibliothèques nécessaires pour votre application avec %pip install <your-library-dependency> afin de garantir qu'elles sont disponibles pour votre cluster Ray et votre application de manière appropriée. La spécification des dépendances dans l'appel de la fonction Ray init installe les dépendances dans un emplacement inaccessible aux nœuds ouvriers Spark, ce qui entraîne des incompatibilités de version et des erreurs d'importation.

Par exemple, vous pouvez exécuter une application Ray simple dans un notebook Azure Databricks comme suit :

import ray
import random
import time
from fractions import Fraction

ray.init()

@ray.remote
def pi4_sample(sample_count):
    """pi4_sample runs sample_count experiments, and returns the
    fraction of time it was inside the circle.
    """
    in_count = 0
    for i in range(sample_count):
        x = random.random()
        y = random.random()
        if x*x + y*y <= 1:
            in_count += 1
    return Fraction(in_count, sample_count)

SAMPLE_COUNT = 1000 * 1000
start = time.time()
future = pi4_sample.remote(sample_count=SAMPLE_COUNT)
pi4 = ray.get(future)
end = time.time()
dur = end - start
print(f'Running {SAMPLE_COUNT} tests took {dur} seconds')

pi = pi4 * 4
print(float(pi))

Créer un cluster Ray en mode de mise à l’échelle automatique

Dans Ray 2.8.0 et versions ultérieures, les clusters Ray démarrés sur Databricks prennent en charge l’intégration à la mise à l’échelle automatique Databricks. Consultez Mise à l’échelle automatique du cluster Databricks.

Avec Ray 2.8.0 et versions ultérieures, vous pouvez créer un cluster Ray sur un cluster Databricks qui prend en charge le scale-up ou le scale-down en fonction des charges de travail. Cette intégration de la mise à l’échelle automatique déclenche la mise à l’échelle automatique du cluster Databricks en interne dans l’environnement Databricks.

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

from ray.util.spark import setup_ray_cluster

setup_ray_cluster(
  num_worker_nodes=8,
  autoscale=True,
  ... # other arguments
)

Si la mise à l’échelle automatique est activée, num_worker_nodes indique le nombre maximal de nœuds Worker Ray. Le nombre minimal par défaut de nœuds Worker Ray est de 0. Ce paramètre par défaut signifie que lorsque le cluster Ray est inactif, il effectuer un scale-down jusqu’à zéro nœud Worker Ray. Cela peut ne pas être idéal pour une réactivité rapide dans tous les scénarios, mais lorsque ce paramètre est activé, cela peut réduire considérablement les coûts.

En mode de mise à l’échelle 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 scale-up and scale-down :

autoscale_upscaling_speed représente le nombre de nœuds autorisés à être en attente sous la forme d’un multiple du nombre actuel de nœuds. Plus la valeur est élevée, plus le scale-up est agressif. Par exemple, si cette valeur est définie sur 1,0, le cluster peut augmenter de taille d’au plus 100 % à tout moment.
autoscale_idle_timeout_minutes représente le nombre de minutes à passer avant qu’un nœud Worker inactif soit supprimé par le générateur de mise à l’échelle automatique. Plus la valeur est faible, plus le scale-down est agressif.

Avec Ray 2.9.0 et versions ultérieures, vous pouvez également définir autoscale_min_worker_nodes pour empêcher le cluster Ray d’effectuer un scale-down à zéro Worker lorsque le cluster Ray est inactif.

Se connecter à un cluster Ray distant à l’aide du client Ray

Dans Ray 2.9.3, créez un cluster Ray en appelant l’API setup_ray_cluster. Dans le même notebook, appelez l’API ray.init() pour vous connecter à ce cluster Ray.

Pour un cluster Ray qui n’est pas en mode global, obtenez la chaîne de connexion distante avec le code suivant :

Pour obtenir la chaîne de connexion distante à l’aide des éléments suivants :

from ray.util.spark import setup_ray_cluster

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

Connectez-vous au cluster distant à l’aide de cette chaîne de connexion distante :

import ray
ray.init(remote_conn_str)

Le client Ray ne prend pas en charge l’API de jeu de données Ray définie dans le module ray.data. Pour contourner ce problème, vous pouvez encapsuler votre code qui appelle l’API de jeu de données Ray à l’intérieur d’une tâche Ray distante, comme indiqué dans le code suivant :

import ray
import pandas as pd
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())

Charger des données dans un DataFrame Spark

Pour charger un DataFrame Spark en tant que jeu de données Ray, vous devez tout d’abord enregistrer le DataFrame Spark dans les volumes UC ou Databricks Filesystem (déconseillé) à l’aide du format Parquet. Pour contrôler l’accès à Databricks Filesystem de manière sécurisée, Databricks recommande de monter le stockage d’objets cloud sur DBFS. Ensuite, vous pouvez créer une instance ray.data.Dataset à partir du chemin du DataFrame Spark sauvegardé en utilisant la méthode auxiliaire suivante :

import ray
import os
from urllib.parse import urlparse

def create_ray_dataset_from_spark_dataframe(spark_dataframe, dbfs_tmp_path):
    spark_dataframe.write.mode('overwrite').parquet(dbfs_tmp_path)
    fuse_path = "/dbfs" + urlparse(dbfs_tmp_path).path
    return ray.data.read_parquet(fuse_path)

# For example, read a Delta Table as a Spark DataFrame
spark_df = spark.read.table("diviner_demo.diviner_pedestrians_data_500")

# Provide a dbfs location to write the table to
data_location_2 = (
    "dbfs:/home/example.user@databricks.com/data/ray_test/test_data_2"
)

# Convert the Spark DataFrame to a Ray dataset
ray_dataset = create_ray_dataset_from_spark_dataframe(
    spark_dataframe=spark_df,
    dbfs_tmp_path=data_location_2
)

Charger des données à partir d’une table Unity Catalog via l’entrepôt Databricks SQL

Pour Ray 2.8.0 et versions ultérieures, vous pouvez appeler l’API ray.data.read_databricks_tables pour charger des données à partir d’une table Databricks Unity Catalog.

Tout d’abord, vous devez définir la variable d’environnement DATABRICKS_TOKEN sur votre jeton d’accès à l’entrepôt Databricks. Si vous n’exécutez pas votre programme sur Databricks Runtime, définissez la variable d’environnement DATABRICKS_HOST sur l’URL de l’espace de travail Databricks, comme indiqué ci-dessous :

export DATABRICKS_HOST=adb-<workspace-id>.<random-number>.azuredatabricks.net

Ensuite, appelez ray.data.read_databricks_tables() pour lire à partir de l’entrepôt SQL Databricks.

import ray

ray_dataset = ray.data.read_databricks_tables(
    warehouse_id='...',  # Databricks SQL warehouse ID
    catalog='catalog_1',  # Unity catalog name
    schema='db_1',  # Schema name
    query="SELECT title, score FROM movie WHERE year >= 1980",
)

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

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

0 cœurs de processeur
0 GPU
128 Mo de mémoire segmentée
128 Mo de mémoire du magasin d’objets

Cela est dû au fait que le nœud principal Ray est généralement utilisé uniquement pour la coordination globale, et non pour l’exécution de tâches Ray. Les ressources du nœud de pilote Spark sont partagées avec plusieurs utilisateurs. Par conséquent, le paramètre par défaut enregistre les ressources côté pilote 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 : définition des cœurs de processeur utilisés par le nœud principal Ray
num_gpus_head_node : définition du GPU utilisé par le nœud principal Ray
object_store_memory_head_node : définition de la taille de mémoire du magasin d’objets par nœud principal Ray

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

Pour des exécutions d’entraînement plus efficaces et plus rentables, vous pouvez créer un cluster Ray sur Spark et définir différentes configurations entre le nœud principal Ray et les nœuds Worker Ray. Toutefois, 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 pilote et de travail en définissant une stratégie 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"
  }
}

Ajuster la configuration du cluster Ray

La configuration recommandée pour chaque nœud Ray Worker est la suivante :

Minimum 4 cœurs CPU par nœud Worker Ray.
Une mémoire heap minimale de 10 Go pour chaque nœud Worker Ray.

Lors de l’appel de ray.util.spark.setup_ray_cluster, Databricks recommande de définir l’argument num_cpus_worker_node sur une valeur >=4.

Consultez la section Allocation de mémoire pour les nœuds ouvriers Ray pour plus de détails sur l'ajustement de la mémoire heap pour chaque nœud Worker Ray.

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

Chaque nœud Worker Ray utilise deux types de mémoire : la mémoire heap 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 ouvriers Ray qui peuvent être lancés sur le nœud Worker Spark. Cela est déterminé par l’argument num_cpus_worker_node ou num_gpus_worker_node.

Si vous ne définissez pas l'argument object_store_memory_per_node, alors la taille de mémoire heap et la taille de 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

De plus, 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 Spark.

Bonnes pratiques

Comment définir le nombre d’UC/GPU pour chaque nœud Worker Ray ?

Databricks recommande de définir num_cpus_worker_node sur le nombre de cœurs d’UC par nœud Worker Spark et num_gpus_worker_node sur le nombre de GPU par nœud Worker Spark. Dans cette configuration, chaque nœud Worker Spark lance un nœud Worker Ray qui utilise entièrement les ressources du nœud Worker Spark.

Configuration du cluster GPU

Le cluster Ray s’exécute sur un cluster Databricks Spark. Un scénario courant consiste à utiliser un travail Spark et une fonction UDF Spark pour effectuer des tâches de prétraitement de données simples qui n’ont pas besoin de ressources GPU, puis utiliser Ray pour exécuter des tâches de Machine Learning complexes qui tirent parti des GPU. Dans ce cas, Databricks recommande de définir le paramètre de configuration au niveau du cluster Spark spark.task.resource.gpu.amount sur 0 pour que toutes les transformations de DataFrame Spark et les exécutions des fonctions UDF Spark n’utilisent pas de ressources GPU.

Les avantages de cette configuration sont les suivants :

Elle augmente le parallélisme des travaux Spark, car le type d’instance GPU a généralement beaucoup plus de cœurs de processeur que les appareils GPU.
Si le cluster Spark est partagé avec plusieurs utilisateurs, cette configuration empêche les travaux Spark de concurrencer les ressources GPU avec des charges de travail Ray en cours d’exécution simultanée.

Désactiver l’intégration `transformers` trainer mlflow si vous l’utilisez dans les tâches Ray

L’intégration transformers trainer MLflow est activée par défaut. Si vous utilisez la formation Ray pour le former, la tâche Ray échoue, car les informations d’identification du service Databricks MLflow ne sont pas configurées pour les tâches Ray.

Pour éviter ce problème, définissez la variable d’environnement DISABLE_MLFLOW_INTEGRATION sur « TRUE » dans la configuration du cluster Databricks. Pour plus d’informations sur la connexion à MLflow dans vos tâches de formation Ray, consultez la section « Utilisation de MLflow dans les tâches Ray ».

Résoudre une erreur de sérialisation (pickling) de la fonction distante Ray

Pour exécuter des tâches Ray, Ray utilise pickle pour sérialiser la fonction de tâche. Si la sérialisation échoue, déterminez la ou les lignes de votre code où l’échec se produit. Souvent, le déplacement de commandes import dans la fonction de tâche résout les erreurs de sérialisation (pickling) courantes. Par exemple, datasets.load_dataset est une fonction largement utilisée qui est patchée dans Databricks Runtime, ce qui peut empêcher la sérialisation (pickling) d’une importation externe. Pour corriger ce problème, vous pouvez mettre à jour votre code comme suit :

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 tuée de manière inattendue avec une erreur OOM

Dans Ray 2.9.3, le moniteur de mémoire Ray présente des problèmes connus qui entraînent la tuerie erronée des tâches Ray.

Pour résoudre le problème, désactivez 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.

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

Si vous exécutez des charges de travail Spark et Ray hybrides dans un cluster Databricks, Databricks vous recommande de réduire la mémoire de l’exécuteur Spark à une petite valeur, en définissant par exemple spark.executor.memory 4g dans la configuration du cluster Databricks. Cela est dû au fait que l’exécuteur Spark s’exécute dans un processus Java qui déclenche le nettoyage de la mémoire (GC) de manière différée. La sollicitation de la mémoire pour la mise en cache des jeux de données Spark est plutôt élevée, ce qui entraîne une réduction de la mémoire disponible que Ray peut utiliser. Pour éviter les erreurs OOM potentielles, Databricks vous recommande de redéfinir la valeur « spark.executor.memory » à une valeur inférieure à la valeur par défaut.

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

Si vous exécutez des charges de travail Spark et Ray hybrides dans un cluster Databricks, définissez les nœuds de cluster Spark sur évolutifs automatiquement, les nœuds Worker Ray sur évolutifs automatiquement ou les deux avec la mise à l’échelle automatique activée.

Par exemple, si vous avez un nombre fixe de nœuds Worker dans un cluster Databricks, envisagez d’activer la mise à l’échelle automatique Ray-on-Spark, afin qu’en l’absence de charge de travail Ray en cours d’exécution, le cluster Ray effectue un scale-down. Par conséquent, les ressources de cluster inactives sont libérées afin que le travail Spark puisse les utiliser.

Une fois le travail Spark terminé et le travail Ray démarré, il déclenche le cluster Ray-on-Spark pour effectuer un scale-up pour répondre aux demandes de traitement.

Vous pouvez également rendre le cluster Databricks et le cluster Ray-on-Spark évolutif automatiquement. Plus précisément, vous pouvez configurer les nœuds évolutifs automatiquement du cluster Databricks sur un maximum de 10 nœuds et les nœuds Worker Ray-on-Spark à un maximum de 4 nœuds (avec un nœud Ray Worker par worker Spark), ce qui permet à Spark d’allouer jusqu’à 6 nœuds pour les tâches Spark. Cela signifie que les charges de travail Ray peuvent utiliser au maximum 4 ressources de nœuds en même temps, tandis que le travail Spark peut allouer au maximum 6 nœuds de ressources.

Application de la fonction de transformation aux lots de données

Lors du traitement des données par lots, Databricks vous recommande d’utiliser l’API Ray Data avec la fonction map_batches. Cette approche peut être plus efficace et évolutive, en particulier pour les jeux de données volumineux ou lors de l’exécution de calculs complexes qui bénéficient du traitement par lots. N’importe quel DataFrame Spark peut être converti en données Ray à l’aide de l’API ray.data.from_spark, et peut être écrit dans la table UC Databricks à l’aide de l’API ray.data.write_databricks_table.

Utilisation de MLflow dans les tâches Ray

Pour utiliser MLflow dans les tâches Ray, configurez les éléments suivants :

Informations d’identification Databricks MLflow dans les tâches Ray
MLflow s’exécute du côté pilote Spark qui transmet les valeurs run_id générées aux tâches Ray.

Le code suivant en est un exemple :

import mlflow
import ray
from mlflow.utils.databricks_utils import get_databricks_env_vars
mlflow_db_creds = get_databricks_env_vars("databricks")

experiment_name = "/Users/<your-name>@databricks.com/mlflow_test"
mlflow.set_experiment(experiment_name)

@ray.remote
def ray_task(x, run_id):
  import os
  os.environ.update(mlflow_db_creds)
  mlflow.set_experiment(experiment_name)
  # We need to use the run created in Spark driver side,
  # and set `nested=True` to make it a nested run inside the
  # parent run.
  with mlflow.start_run(run_id=run_id, nested=True):
    mlflow.log_metric(f"task_{x}_metric", x)
  return x

with mlflow.start_run() as run:  # create MLflow run in Spark driver side.
  results = ray.get([ray_task.remote(x, run.info.run_id) for x in range(10)])

Utilisation de bibliothèques Python étendues au notebook ou de bibliothèques python de cluster dans les tâches Ray

Actuellement, Ray a un problème connu où les tâches Ray ne peuvent pas utiliser les bibliothèques Python délimitées aux notebooks ou les bibliothèques Python de cluster. Pour résoudre cette limitation, exécutez la commande suivante dans votre notebook avant de lancer un cluster Ray-on-Spark :

%pip install ray==<The Ray version you want to use> --force-reinstall

puis exécutez la commande suivante dans votre notebook pour redémarrer le noyau Python :

dbutils.library.restartPython()

Activer les traces de la pile et les graphiques enflammés sur la page des acteurs du tableau de bord Ray.

Sur la page des acteurs du tableau de bord Ray, vous pouvez visualiser les traces de la pile et les graphiques enflammés pour les acteurs Ray actifs.

Pour afficher ces informations, installez py-spy avant de démarrer le cluster Ray :

%pip install py-spy

Arrêter un cluster Ray

Pour arrêter un cluster Ray en cours d’exécution sur Azure Databricks, appelez l’API ray.utils.spark.shutdown_ray_cluster.