Aller au contenu principal
Dernière mise à jour le

Charger des données sur AI Runtime

info

Aperçu

Cette fonctionnalité est en aperçu public.

Chargez les données d'entraînement sur le Runtime d'IA pour les charges de travail de Machine Learning et de deep learning. Tout accès aux données passe par Unity Catalog : utilisez Spark Connect pour lire les données tabulaires des tables Delta, et les volumes Unity Catalog pour les grands datasets et les fichiers non structurés tels que les images, l'audio et le texte. Pour l'entraînement multi-époques, mettez les données en cache localement dans /tmp pour un accès plus rapide. Pour apprendre à charger et à transformer des données avec l'API Python de Spark, consultez le didacticiel.

remarque

Unity Catalog est requis. Tout accès aux données sur AI Runtime passe par Unity Catalog. Vos tables et volumes doivent être enregistrés dans Unity Catalog et accessibles à votre utilisateur ou à votre Service Principal.

Charger des données tabulaires​

Utilisez Spark Connect pour charger des données de Machine Learning tabulaires à partir de tables Delta.

Pour l'entraînement sur nœud unique, vous pouvez convertir les DataFrames Apache Spark en DataFrames Pandas à l'aide de la méthode PySpark toPandas(), puis, en option, les convertir au format NumPy à l'aide de la méthode PySpark to_numpy().

remarque

Spark Connect diffère l'analyse et la résolution de noms à la durée d'exécution, ce qui peut modifier le comportement de votre code. Voir Comparaison de Spark Connect à Spark Classic.

Spark Connect prend en charge la plupart des API PySpark, y compris Spark SQL, l'API Pandas sur Spark, Structured Streaming et MLlib (basé sur les DataFrames). Consultez la documentation de référence de l'API PySpark pour les dernières APIs prises en charge.

Pour les autres limitations, consultez les limitations du compute Serverless.

Chargez de grandes tables Delta à l’aide de volumes​

Pour les grandes tables Delta qui sont trop grandes pour être converties avec toPandas(), exportez les données vers un volume Unity Catalog et chargez-les directement à l'aide de PyTorch ou de Hugging Face :

Python
# Step 1: Export the Delta table to Parquet files in a UC volume
output_path = "/Volumes/catalog/schema/my_volume/training_data"
spark.table("catalog.schema.my_table").write.mode("overwrite").parquet(output_path)
Python
# Step 2: Load the exported data directly using Hugging Face datasets
from datasets import load_dataset

dataset = load_dataset("parquet", data_files="/Volumes/catalog/schema/my_volume/training_data/*.parquet")

Cette approche évite le surcoût de Spark pendant l'entraînement et fonctionne bien pour les workflows d'entraînement à GPU unique et distribués.

Charger des données non structurées à partir de volumes avec UCVolumeDataset​

Pour les données non structurées telles que les images, les fichiers audio et texte stockés dans les volumes Unity Catalog, utilisez UCVolumeDataset du package serverless_gpu.data. UCVolumeDataset est un IterableDataset PyTorch qui copie chaque fichier du volume vers un cache local rapide lors du premier accès et renvoie le chemin d’accès au fichier local mis en cache. Il gère les problèmes de performances et de distribution que vous mettriez en œuvre à la main :

  • Mise en cache locale. Les fichiers sont copiés du montage FUSE vers un répertoire de cache local lors du premier accès et servis à partir du cache par la suite, de sorte que l'entraînement multi-époque ne relit pas le volume.
  • **Partitionnement automatique.** Lorsque torch.distributed est initialisé, les fichiers sont partitionnés entre les rangs, puis divisés entre DataLoader Workers, de sorte que chaque paire (rank, worker) reçoit une tranche non superposée sans configuration supplémentaire.
remarque

UCVolumeDataset et serverless_gpu.data.DataLoader nécessitent un environnement GPU 5 ou supérieur.

UCVolumeDataset produit des chemins de fichiers locaux bruts. Pour décoder ces fichiers en tenseurs, enveloppez-le dans un deuxième IterableDataset qui consomme le Stream de chemin et applique votre logique d'analyse. Cela permet de séparer les préoccupations liées aux E/S et à l'analyse.

Python
from serverless_gpu.data import UCVolumeDataset
from torch.utils.data import IterableDataset
from PIL import Image
import torchvision.transforms.functional as TF

class ImageDataset(IterableDataset):
    """Decodes each cached file path from UCVolumeDataset into a tensor."""

    def __init__(self, path_dataset: UCVolumeDataset):
        self._path_dataset = path_dataset

    def __iter__(self):
        for local_path in self._path_dataset:
            image = Image.open(local_path).convert("RGB")
            yield TF.to_tensor(image)

path_dataset = UCVolumeDataset("/Volumes/catalog/schema/my_volume/images")
dataset = ImageDataset(path_dataset)

Le wrapper reçoit des chemins locaux déjà mis en cache, de sorte que l’étape d’analyse ne touche jamais le montage FUSE. Vous pouvez chaîner des wrappers supplémentaires pour l’augmentation, la tokenisation ou le filtrage.

Pour des performances optimales, associez UCVolumeDataset à serverless_gpu.data.DataLoader plutôt qu'avec le PyTorch standard DataLoader. Il est optimisé pour les E/S GPU serverless et récupère et met en cache les fichiers simultanément pendant que le GPU effectue les calculs. Découvrez les performances de chargement des données.

Charger les données à l'intérieur du décorateur @distributed​

Lorsque vous utilisez l'API GPU Serverless pour l'entraînement distribué, déplacez le code de chargement de données à l'intérieur du décorateur @distributed. La taille du dataset peut dépasser la taille maximale autorisée par pickle, il est donc recommandé de générer le dataset à l'intérieur du décorateur, comme indiqué ci-dessous :

Python
from serverless_gpu import distributed

# This may cause a pickle error if the dataset is too large
dataset = get_dataset(file_path)

@distributed(gpus=8, gpu_type='H100')
def run_train():
    # Load data inside the decorator to avoid pickle serialization issues
    dataset = get_dataset(file_path)
    ...

Lorsque vous construisez un UCVolumeDataset à l'intérieur du décorateur, il lit les informations de rang torch.distributed au moment de l'itération et partitionne les fichiers entre les rangs automatiquement, vous n'avez donc pas besoin d'un DistributedSampler pour les données de volume basées sur les fichiers.

Performances de chargement des données​

/Workspace et les répertoires /Volumes sont hébergés sur un stockage Unity Catalog distant. Si votre dataset est stocké dans Unity Catalog, la vitesse de chargement des données est limitée par la bande passante réseau disponible. Si vous entraînez plusieurs époques, l'approche recommandée est d'utiliser UCVolumeDataset qui effectue cette mise en cache pour vous : il copie chaque fichier dans le stockage local lors du premier accès et sert les lectures ultérieures à partir de la copie locale. Pour les datasets dans les volumes, préférez-le à un shutil.copytree manuel, qui copie l'intégralité de l'arborescence à l'avance même si l'entraînement n'en touche qu'une partie.

Si votre dataset est volumineux, les techniques suivantes peuvent améliorer le throughput :

  • Utilisez serverless_gpu.data.DataLoader pour paralléliser la récupération. Ceci est une sous-classe remplaçable du torch DataLoader optimisée pour l'E/S GPU Serverless : num_workers prend par default la valeur 6 et prefetch_factor la valeur 4 (par rapport à 0 et 2 de PyTorch), les fichiers sont donc récupérés et mis en cache simultanément pendant que le GPU calcule. Il enregistre également le temps de récupération par batch pour l'exécution MLflow active, ce qui vous aide à repérer les goulots d'étranglement liés au chargement des données.

    Python
    from serverless_gpu.data import DataLoader

    loader = DataLoader(
        dataset,
        batch_size=32,
        pin_memory=True,
        # num_workers=6, by default
        # prefetch_factor=4, by default
        # raise num_workers to increase parallel reads, or prefetch_factor to deepen each worker's queue.
    )

    Tous les rangs doivent utiliser la même valeur num_workers, car UCVolumeDataset partitionne les fichiers en utilisant un pas global sur world_size × num_workers emplacements. Des valeurs incohérentes entraînent la duplication ou l'omission de fichiers.

  • **Augmentez la taille du batch.** Des batches plus grands amortissent la surcharge de chargement de données par batch sur un plus grand nombre d'échantillons et réduisent le nombre d'opérations de récupération de fichiers par étape. Si la mémoire GPU est le facteur limitant, combinez une taille de batch plus grande avec l'accumulation de gradient pour préserver la taille de batch effective.

Jeux de données de streaming​

Pour les très grands datasets qui ne tiennent pas en mémoire, utilisez les approches de streaming :

Sur cette page