Aller au contenu principal
Dernière mise à jour le

Charger les données à l'aide de Mosaic Streaming

Cet article explique comment utiliser Mosaic Streaming pour convertir les données d'Apache Spark dans un format compatible avec PyTorch.

Mosaic Streaming est une bibliothèque open source de chargement de données. Il permet l'entraînement et l'évaluation en nœud unique ou distribués des modèles de deep learning à partir de datasets déjà chargés en tant que DataFrames Apache Spark. Mosaic Streaming prend principalement en charge Mosaic Composer, mais s’intègre également à PyTorch natif, PyTorch Lightning et TorchDistributor. Mosaic Streaming offre une série d'avantages par rapport aux DataLoaders PyTorch traditionnels, notamment :

  • Compatibilité avec tout type de données, y compris les images, le texte, la vidéo et les données multimodales.
  • Prise en charge des principaux fournisseurs de stockage cloud (AWS, OCI, GCS, Azure, Databricks UC Volume et tout stockage d'objets compatible S3 comme Cloudflare R2, Coreweave, Backblaze b2, etc.).
  • Optimisation des garanties d'exactitude, des performances, de la flexibilité et de la facilité d'utilisation. Pour plus d'informations, consultez leur page fonctionnalités clés.

Pour des informations générales sur Mosaic Streaming, consultez la documentation de l’API Streaming.

remarque

Mosaic Streaming a été préinstallé dans toutes les versions du Databricks Runtime 15.2 ML et supérieures.

Charger des données depuis des DataFrames Spark à l'aide de Mosaic Streaming​

Mosaic Streaming offre un workflow simple pour la conversion d'Apache Spark au format Mosaic Data Shard (MDS), qui peut ensuite être chargé pour être utilisé dans un environnement distribué.

Le workflow recommandé est :

  1. Utilisez Apache Spark pour charger et éventuellement prétraiter les données.
  2. Utilisez streaming.base.converters.dataframe_to_mds pour enregistrer le dataframe sur le disque pour un stockage temporaire et/ou dans un volume Unity Catalog pour un stockage persistant. Ces données seront stockées au format MDS et pourront être optimisées grâce à la prise en charge de la compression et du hachage. Les cas d'utilisation avancés peuvent également inclure le prétraitement des données à l'aide de fonctions UDF. Consultez le didacticiel Spark DataFrame vers MDS pour plus d'information.
  3. Utilisez streaming.StreamingDataset pour charger les données nécessaires en mémoire. StreamingDataset est une version d'IterableDataset de PyTorch qui propose un brassage élastiquement déterministe, ce qui permet une reprise rapide à mi-époque. Consultez la documentation StreamingDataset pour plus d'informations.
  4. Utilisez streaming.StreamingDataLoader pour charger les données nécessaires à la formation/l'évaluation/au test. StreamingDataLoader est une version du DataLoader de PyTorch qui fournit une interface supplémentaire de point de contrôle/reprise, pour laquelle il suit le nombre d'échantillons vus par le modèle dans ce rang.

Pour un exemple de bout en bout, voir le notebook suivant :

Simplifiez le chargement de données de Spark vers PyTorch à l’aide du Notebook Mosaic Streaming

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Dépannage​

Erreur d'authentification​

Si vous rencontrez l'erreur suivante lors du chargement de données à partir d'un volume Unity Catalog à l'aide de StreamingDataset, configurez les variables d'environnement comme indiqué ci-dessous.

sh
ValueError: default auth: cannot configure default credentials, please check https://docs.databricks.com/en/dev-tools/auth.html#databricks-client-unified-authentication to configure credentials for your preferred authentication method.
remarque

Si vous rencontrez cette erreur lors de l'exécution de l'entraînement distribué à l'aide de TorchDistributor, vous devez également définir les variables d'environnement sur les nœuds worker.

Python
db_host = "https://your-databricks-host.databricks.com"
db_token = "YOUR API TOKEN" # Create a token with either method from https://docs.databricks.com/en/dev-tools/auth/index.html#databricks-authentication-methods

def your_training_function():
  import os
  os.environ['DATABRICKS_HOST'] = db_host
  os.environ['DATABRICKS_TOKEN'] = db_token

# The above function can be distributed with TorchDistributor:
# from pyspark.ml.torch.distributor import TorchDistributor
# distributor = TorchDistributor(...)
# distributor.run(your_training_function)

Problèmes de mémoire partagée de Python 3.11​

En raison de problèmes liés à l’implémentation de la mémoire partagée de Python 3.11, StreamingDataset peut rencontrer des problèmes transitoires sur Databricks Runtime 15.4 LTS pour Machine Learning. Vous pouvez éviter ces problèmes en passant à Databricks Runtime 16.4 LTS pour Machine Learning, car Python 3.12 les résout.

Sur cette page