Ładowanie danych w środowisku uruchomieniowym sztucznej inteligencji

Ważna

Ta funkcja jest dostępna w publicznej wersji testowej.

Ładowanie danych treningowych w środowisku AI Runtime na potrzeby obciążeń uczenia maszynowego i uczenia głębokiego. Cały dostęp do danych odbywa się za pośrednictwem Unity Catalog: użyj Spark Connect do odczytu danych tabelarycznych z tabel Delta, a woluminów Unity Catalog do pracy z dużymi zbiorami danych i plikami nieustrukturyzowanymi, takimi jak obrazy, dźwięk i tekst. W przypadku trenowania wieloepokowego zapisuj dane lokalnie w pamięci podręcznej do /tmp, aby uzyskać szybszy dostęp. Aby dowiedzieć się, jak ładować i przekształcać dane za pomocą interfejsu API platformy Spark Python, zobacz samouczek.

Uwaga / Notatka

Unity Catalog jest wymagany. Cały dostęp do danych w AI Runtime odbywa się poprzez Unity Catalog. Tabele i zasoby muszą być zarejestrowane w katalogu Unity i dostępne dla użytkownika lub podmiotu usługowego.

Ładowanie danych tabelarycznych

Użyj programu Spark Connect, aby załadować tabelaryczne dane uczenia maszynowego z tabel delta.

W przypadku trenowania z jednym węzłem można przekonwertować ramki danych platformy Apache Spark na ramki danych biblioteki pandas przy użyciu metody toPandas(), a następnie opcjonalnie przekonwertować na format NumPy przy użyciu metody to_numpy().

Uwaga / Notatka

Spark Connect odracza analizę i rozpoznawanie nazw do czasu kompilacji, co może wpłynąć na sposób działania twojego kodu. Zobacz Porównanie programu Spark Connect z modelem klasycznym platformy Spark.

Narzędzie Spark Connect obsługuje większość interfejsów API PySpark, w tym Spark SQL, interfejs API Pandas na platformie Spark, uporządkowane przesyłanie strumieniowe oraz bibliotekę MLlib (opartą na ramkach danych). Aby uzyskać najnowsze obsługiwane interfejsy API, zobacz dokumentację interfejsu API PySpark .

Aby uzyskać informacje o innych ograniczeniach, zobacz Ograniczenia obliczeniowe bezserwerowe.

Ładowanie dużych tabel Delta za pomocą woluminów

W przypadku dużych tabel Delta, które są zbyt duże do konwersji za pomocą toPandas(), wyeksportuj dane do woluminu Unity Catalog i załaduj je bezpośrednio z użyciem PyTorch lub Hugging Face.

# 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)

# 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")

Takie podejście pozwala uniknąć narzutów platformy Spark podczas trenowania i działa dobrze zarówno w przypadku procesów trenowania pojedynczego GPU, jak i rozproszonej pracy trenowania.

Ładowanie danych bez struktury z woluminów za pomocą polecenia UCVolumeDataset

W przypadku danych nieustrukturyzowanych, takich jak obrazy, pliki audio i pliki tekstowe przechowywane w woluminach Unity Catalog, użyj UCVolumeDataset z pakietu serverless_gpu.data. UCVolumeDataset to element PyTorch IterableDataset, który przy pierwszym dostępie kopiuje każdy plik z woluminu do szybkiej lokalnej pamięci podręcznej i zwraca ścieżkę do zbuforowanego lokalnego pliku. Obsługuje ona problemy z wydajnością i dystrybucją, które w przeciwnym razie należy zaimplementować ręcznie:

  • Buforowanie lokalne. Pliki są kopiowane z punktu montowania FUSE do lokalnego katalogu pamięci podręcznej przy pierwszym dostępie, a następnie udostępniane z pamięci podręcznej, więc trening wieloepokowy nie powoduje ponownego odczytu woluminu.
  • Automatyczne partycjonowanie. Po zainicjowaniu torch.distributed pliki są dzielone między rangi, a następnie między procesy robocze DataLoader, dzięki czemu każda para (rank, worker) otrzymuje niepokrywający się fragment bez dodatkowej konfiguracji.

Uwaga / Notatka

UCVolumeDataset i serverless_gpu.data.DataLoader wymagają środowiska procesora GPU 5 lub nowszego.

UCVolumeDataset zwraca nieprzetworzone ścieżki plików lokalnych. Aby zdekodować te pliki do tensorów, opakuj je w drugi element IterableDataset, który korzysta ze strumienia ścieżek i stosuje logikę parsowania. Dzięki temu kwestie związane z I/O i parsowaniem pozostają oddzielone.

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)

Warstwa pośrednia otrzymuje ścieżki lokalne wcześniej zapisane w pamięci podręcznej, więc etap parsowania nigdy nie odwołuje się do punktu montowania FUSE. Możesz łączyć w łańcuch dodatkowe opakowania do rozszerzania, tokenizacji lub filtrowania.

Aby uzyskać optymalną wydajność, używaj UCVolumeDataset w połączeniu z serverless_gpu.data.DataLoader, zamiast domyślnego DataLoader z PyTorch. Jest on dostrojony dla bezserwerowych operacji we/wy procesora GPU i pobiera i buforuje pliki współbieżnie podczas obliczeń procesora GPU. Zobacz Wydajność ładowania danych.

Ładowanie danych wewnątrz dekoratora @distributed

Podczas korzystania z Serverless GPU API do rozproszonego trenowania, umieść kod ładowania danych wewnątrz dekoratora @distributed. Rozmiar zestawu danych może przekraczać maksymalny rozmiar dozwolony przez pickle, dlatego zaleca się wygenerować zestaw danych wewnątrz dekoratora, co przedstawiono poniżej.

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)
    ...

Podczas konstruowania UCVolumeDataset wewnątrz dekoratora to odczytuje torch.distributed informacje o rangach w czasie iteracji i automatycznie partycjonuje pliki pomiędzy rangi, więc DistributedSampler nie jest potrzebny do danych woluminowych opartych na plikach.

Wydajność ładowania danych

/Workspace i /Volumes katalogi są hostowane na zdalnym magazynie Unity Catalog. Jeśli zestaw danych jest przechowywany w Unity Catalog, szybkość ładowania danych jest ograniczona przez dostępną szerokość pasma sieci. Jeśli trenujesz przez wiele epok, zalecanym podejściem jest użycie UCVolumeDataset, które realizuje to buforowanie: przy pierwszym dostępie kopiuje każdy plik do pamięci lokalnej, a kolejne odczyty obsługuje z kopii lokalnej. W przypadku zestawów danych w woluminach należy preferować tę opcję zamiast ręcznego shutil.copytree, które z wyprzedzeniem kopiuje całe drzewo, nawet jeśli podczas treningu używana jest tylko jego część.

Jeśli zestaw danych jest duży, następujące techniki mogą zwiększyć przepływność:

  • Użyj serverless_gpu.data.DataLoader, aby zrównoleglić pobieranie. Jest to podklasa typu drop-in klasy torch DataLoader, dostrojona do bezserwerowych operacji wejścia/wyjścia GPU: num_workers ma domyślnie wartość 6, a prefetch_factor — 4 (w porównaniu z wartościami 0 i 2 w PyTorch), dzięki czemu pliki są pobierane i buforowane współbieżnie, gdy GPU wykonuje obliczenia. Rejestruje również chronometraż pobierania poszczególnych partii do aktywnego przebiegu platformy MLflow, co ułatwia wykrycie wąskich gardeł ładowania danych.

    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.
)

    Wszystkie rangi muszą używać tej samej wartości num_workers, ponieważ UCVolumeDataset partycjonuje pliki przy użyciu globalnego kroku między gniazdami world_size × num_workers. Niedopasowane wartości powodują duplikowanie lub pomijanie plików.

  • Zwiększ rozmiar partii. Większe partie rozkładają narzut związany z ładowaniem danych dla każdej partii na większą liczbę próbek i zmniejszają liczbę operacji pobierania plików w każdym kroku. Jeśli pamięć GPU jest czynnikiem ograniczającym, połącz większy rozmiar partii z akumulacją gradientu, aby zachować efektywny rozmiar partii.

Zestawy danych przesyłania strumieniowego

W przypadku bardzo dużych zestawów danych, które nie mieszczą się w pamięci, użyj metod przesyłania strumieniowego:

  • Last updated on