Aller au contenu principal
Dernière mise à jour le

Combinez Ray et Spark dans le même environnement sur Databricks

Avec Databricks, vous pouvez exécuter des opérations Ray et Spark dans le même environnement d'exécution pour tirer parti des atouts des deux moteurs de calcul distribué.

L'intégration de Ray et Spark est prise en charge par Delta Lake et Unity Catalog, qui offrent une gestion de données robuste, un accès sécurisé et le suivi de la traçabilité.

Cet article vous montre comment connecter les opérations Ray et Spark selon les cas d'utilisation suivants :

  • Écrire des données Spark vers des données Ray : transférer efficacement des données en mémoire vers Ray.
  • Écrire les données Ray dans Spark : produisez les données de Ray vers Delta Lake ou d'autres solutions de stockage pour garantir la compatibilité et l'accès.
  • Connecter des applications Ray externes à Unity Catalog : Connectez les applications Ray en dehors de Databricks pour charger des données d'une table Unity Catalog Databricks.

Pour plus d'informations sur l'utilisation de Ray, Spark ou des deux, consultez Quand utiliser Spark ou Ray.

Créez un dataset Ray distribué à partir d’un DataFrame Spark​

Pour créer un dataset Ray distribué à partir d'un Spark DataFrame, vous pouvez utiliser la fonction ray.data.from_spark() pour lire directement un Spark DataFrame depuis Ray sans avoir besoin d'écrire les données à un quelconque emplacement.

Les transferts Spark vers Ray en mémoire sont disponibles sur Databricks Runtime ML 15.0 et versions supérieures.

Pour activer cette fonctionnalité, vous devez procéder comme suit :

  • Définissez la configuration du cluster Spark spark.databricks.pyspark.dataFrameChunk.enabled sur true avant de démarrer votre cluster.
Python
import ray.data

source_table = "my_db.my_table"

# Read a Spark DataFrame from a Delta table in Unity Catalog
df = spark.read.table(source_table)
ray_ds = ray.data.from_spark(df)
attention

Les clusters Spark à mise à l'échelle automatique (y compris ceux qui utilisent des instances spot) doivent définir le parameter use_spark_chunk_api sur False pour utiliser la fonction from_spark(). Dans le cas contraire, l'appel d'API entraînera des échecs de cache car le cache sur un exécuteur Spark est perdu lorsque l'exécuteur se termine.

Python
ray_ds = ray.data.from_spark(df, use_spark_chunk_api=False)

Écrire les données Ray vers Spark​

Pour écrire les données Ray vers Spark, vous devez écrire le dataset dans un emplacement auquel Spark peut accéder.

Dans Databricks Runtime ML inférieur à 15.0, vous pouvez écrire directement dans un emplacement de stockage d'objets à l'aide de l'enregistreur Parquet de Ray, ray_dataset.write_parquet() du module ray.data. Spark peut lire ces données Parquet avec des lecteurs natifs.

Pour les workspaces où Unity Catalog est activé, utilisez la fonction ray.data.Dataset.write_databricks_table pour écrire dans une table Unity Catalog.

Cette fonction stocke temporairement le dataset Ray dans les volumes Unity Catalog, lit les volumes Unity Catalog avec Spark, et écrit enfin dans une table Unity Catalog. Avant d'appeler la fonction ray.data.Dataset.write_databricks_table, assurez-vous que la variable d'environnement "RAY_UC_VOLUMES_FUSE_TEMP_DIR" est définie sur un chemin de volume Unity Catalog valide et accessible, comme "/Volumes/MyCatalog/MySchema/MyVolume/MyRayData".

Python
ds = ray.data
ds.write_databricks_table()

Pour les Workspaces pour lesquels Unity Catalog n'est pas activé, vous pouvez stocker manuellement un dataset Ray Data en tant que fichier temporaire, tel qu'un fichier Parquet dans DBFS, puis lire le fichier de données avec Spark.

Python
ds.write_parquet(tmp_path)
df = spark.read.parquet(tmp_path)
df.write.format("delta").saveAsTable(table_name)

Écrire des données depuis les applications Ray core vers Spark​

Databricks peut également intégrer des applications Ray Core avec Spark, ce qui vous permet d’exécuter des charges de travail Ray Core (les APIs de niveau inférieur de Ray) et Spark dans le même environnement et de faciliter l’échange de données entre eux. Cette intégration offre plusieurs modèles pour s’adapter aux différentes charges de travail et aux besoins en gestion de données, garantissant une expérience simplifiée à l’aide des deux frameworks.

Il existe trois principaux modèles pour écrire des données de Ray vers Spark.

  • Conserver la sortie dans un emplacement temporaire : stockez temporairement les sorties de tâches Ray dans des volumes DBFS ou Unity Catalog avant de les consolider dans un Spark DataFrame.
  • Connecter avec Spark Connect : Connectez directement les tâches Ray à un cluster Spark, permettant à Ray d'interagir avec les DataFrames et les tables Spark.
  • Utiliser des bibliothèques tierces : utilisez des bibliothèques externes, telles que deltalake ou deltaray, pour écrire des données des tâches Ray Core dans des tables Delta Lake ou Spark.

Modèle 1 : conservez la sortie dans un emplacement temporaire​

Le modèle le plus courant pour écrire des données de Ray vers Spark consiste à stocker les données de sortie dans un emplacement temporaire, par exemple des volumes Unity Catalog ou DBFS. Après le stockage des données, le thread du driver Ray lit chaque partie des fichiers sur les nœuds worker et les consolide dans un DataFrame final pour un traitement ultérieur. En général, les fichiers temporaires sont dans un format standard comme CSV. Cette approche fonctionne le mieux lorsque les données de sortie sont sous forme tabulaire, comme un DataFrame Pandas généré par une tâche Ray Core.

Utilisez cette méthode lorsque la sortie des tâches Ray est trop volumineuse pour tenir dans la mémoire du nœud Driver ou dans le magasin d’objets partagés. Si vous devez gérer des datasets volumineux sans persister les données sur le stockage, envisagez d'augmenter la mémoire allouée au nœud Driver dans votre cluster Databricks pour améliorer les performances.

Python
import os
import uuid
import numpy as np
import pandas as pd

@ray.remote
def write_example(task_id, path_prefix):

  num_rows = 100

  df = pd.DataFrame({
      'foo': np.random.rand(num_rows),
      'bar': np.random.rand(num_rows)
  })

  # Write the DataFrame to a CSV file
  df.to_csv(os.path.join(path_prefix, f"result_part_{task_id}.csv"))

n_tasks = 10

# Put a unique Unity Catalog volume prefix for the temporary file path
volume_prefix = f"/Volumes/<catalog>/<schema>/<volume>/<USERNAME>"

# Create a unique path for the temporary files
path_prefix = os.path.join(volume_prefix, f"ray_tmp/write_task_{uuid.uuid4()}")

tasks = ray.get([write_example.remote(i, path_prefix) for i in range(n_tasks)])

# Read all CSV files in the directory into a single DataFrame
df = spark.read.csv(path_prefix, header=True, inferSchema=True)

Modèle 2 : Connexion à l'aide de Spark Connect​

Une autre façon pour les tâches Ray Core d'interagir avec Spark au sein de la tâche distante est d'utiliser Spark Connect. Cela vous permet de configurer le contexte Spark sur le worker Ray pour pointer vers le cluster Spark s'exécutant à partir du nœud driver.

Pour ce faire, vous devez configurer les ressources du cluster Ray afin d'allouer de l'espace à Spark. Par exemple, si un nœud worker a 8 CPU, définissez num_cpus_worker_node sur 7, en laissant 1 CPU pour Spark. Pour les tâches Spark plus importantes, il est recommandé d'allouer une plus grande part des ressources.

Python
from databricks.connect import DatabricksSession
import ray

@ray.remote
class SparkHandler(object):

   def __init__(self, access_token=None, cluster_id=None, host_url=None):
       self.spark = (DatabricksSession
                     .builder
                     .remote(host=host_url,
                             token=access_token,
                             cluster_id=cluster_id)
                     .getOrCreate()
                     )
   def test(self):
       df = self.spark.sql("select * from samples.nyctaxi.trips")

       df.write.format("delta").mode(
"overwrite").saveAsTable("catalog.schema.taxi_trips")
       return df.count()

access_token = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get()
cluster_id = dbutils.notebook.entry_point.getDbutils().notebook().getContext().clusterId().get()
host_url = f"https://{dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().get('browserHostName').get()}"

sh = SparkHandler.remote(access_token=access_token,
                        cluster_id=cluster_id,
                        host_url=host_url)
print(ray.get(sh.test.remote()))

Cet exemple utilise le jeton généré par le Notebook. Cependant, Databricks recommande que les cas d'utilisation en production utilisent un jeton d'accès stocké dans les secrets Databricks.

Étant donné que ce processus appelle un seul driver Spark, il crée un *verrou de thread* qui oblige toutes les tâches à attendre la fin des tâches Spark précédentes. Par conséquent, il est recommandé de l'utiliser lorsqu'il n'y a pas beaucoup de tâches concurrentes, car elles auront toutes un comportement séquentiel à mesure que les tâches Spark se terminent. Pour ces situations, il est préférable de conserver la sortie, puis de la combiner dans un seul dataframe Spark à la fin, puis de l'écrire dans une table de sortie.

Modèle 3 : Bibliothèques tierces​

Une autre option consiste à utiliser des bibliothèques tierces qui interagissent avec Delta Lake et Spark. Databricks ne prend pas officiellement en charge ces bibliothèques tierces. Un exemple de ceci est la bibliothèque deltalake du projet delta-rs. Cette approche ne fonctionne actuellement qu'avec les tables Hive metastore, et non avec les tables Unity Catalog.

Python
from deltalake import DeltaTable, write_deltalake
import pandas as pd
import numpy as np
import ray

@ray.remote
def write_test(table_name):
   random_df_id_vals = [int(np.random.randint(1000)), int(np.random.randint(1000))]
   pdf = pd.DataFrame({"id": random_df_id_vals, "value": ["foo", "bar"]})
   write_deltalake(table_name, pdf, mode="append")

def main():
   table_name = "database.mytable"
   ray.get([write_test.remote(table_name) for _ in range(100)])

Une autre bibliothèque tierce disponible est la bibliothèque deltaray, accessible via le projet Delta Incubator https://github.com/delta-incubator/deltaray).

Python
# Standard Libraries
import pathlib

# External Libraries
import deltaray
import deltalake as dl
import pandas as pd

# Creating a Delta Table
cwd = pathlib.Path().resolve()
table_uri = f'{cwd}/tmp/delta-table'
df = pd.DataFrame({'id': [0, 1, 2, 3, 4, ], })
dl.write_deltalake(table_uri, df)

# Reading our Delta Table
ds = deltaray.read_delta(table_uri)
ds.show()

Connectez les applications Ray externes à Databricks​

Créer un dataset Ray à partir d’une query de Databricks warehouse​

Pour Ray 2.8.0 et versions supérieures, pour connecter des applications Ray en dehors de Databricks à des tables dans Databricks, vous pouvez appeler l'API ray.data.read_databricks_tables pour charger des données depuis une table Unity Catalog.

Tout d'abord, définissez la variable d'environnement DATABRICKS_TOKEN sur votre jeton d'accès au SQL warehouse. Si vous n'exécutez pas votre programme sur Databricks Runtime, définissez également la variable d'environnement DATABRICKS_HOST sur l'URL du workspace Databricks, comme indiqué ci-après :

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

Ensuite, appelez ray.data.read_databricks_tables() pour lire depuis le SQL warehouse.

Python
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",
)
attention

Les Databricks warehouses ne peuvent mettre en cache les résultats de query que pendant environ 2 heures. Pour les charges de travail de longue durée, appelez la méthode ray.data.Dataset.materialize afin de matérialiser le dataset Ray dans le magasin d'objets distribués Ray.

Créer un dataset Ray à partir d’une table Databricks OpenSharing​

Vous pouvez également lire des données à partir des tables OpenSharing de Databricks. La lecture à partir des tables OpenSharing est plus fiable que la lecture à partir d'un cache de Databricks warehouse.

L'API ray.data.read_delta_sharing_tables est disponible sur Ray 2.33 et versions ultérieures.

Python
import ray

ds = ray.data.read_delta_sharing_tables(
    url=f"<profile-file-path>#<share-name>.<schema-name>.<table-name>",
    limit=100000,
    version=1,
)

Bonnes pratiques​

  • Utilisez toujours les techniques décrites dans le guide des bonnes pratiques du cluster Ray afin de garantir que le cluster est pleinement utilisé.
  • Envisagez d'utiliser les volumes Unity Catalog pour stocker les données de sortie dans un format non tabulaire et assurer la gouvernance.
  • Assurez-vous que la configuration num_cpus_worker_node est définie de sorte que le nombre de cœurs de processeur corresponde à celui du nœud worker Spark. De même, définissez num_gpus_worker_node comme 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 pleinement les ressources du nœud worker Spark.

Limitations​

Unity Catalog ne partage actuellement pas les identifiants pour l'écriture dans les tables à partir de rédacteurs non-Spark. Par conséquent, toutes les données écrites dans une table Unity Catalog à partir d'une tâche Ray Core nécessiteront que les données soient persistées, puis lues avec Spark, ou que Databricks Connect soit configuré au sein de la tâche Ray.

Sur cette page