Загрузка данных в среде выполнения ИИ

Это важно

Среда выполнения ИИ для задач с одним узлом находится в общедоступной предварительной версии. API распределенного обучения для рабочих нагрузок с несколькими GPU остаются в бета-версии.

В этом разделе рассматриваются сведения о загрузке данных в среду выполнения ИИ, специально предназначенную для приложений машинного и глубокого обучения. Ознакомьтесь с руководством , чтобы узнать больше о том, как загружать и преобразовывать данные с помощью API Python Spark.

Замечание

Требуется каталог Unity. Все доступ к данным в среде выполнения ИИ проходит через каталог Unity. Ваши таблицы и тома должны быть зарегистрированы в Unity Catalog и доступны вашим пользователю или учетной записи службы.

Загрузка табличных данных

Используйте Spark Connect для загрузки табличных данных машинного обучения из таблиц Delta.

Для обучения с одним узлом можно преобразовать Apache Spark DataFrames в pandas DataFrames с помощью метода PySpark, а затем при необходимости преобразовать в формат NumPy с помощью метода toPandas()to_numpy().

Замечание

Spark Connect откладывает анализ и разрешение имен во время выполнения, что может изменить поведение кода. См . статью "Сравнение Spark Connect с классической версией Spark".

Spark Connect поддерживает большинство API PySpark, включая Spark SQL, Pandas API в Spark, Structured Streaming и MLlib (на базе DataFrame). См. справочную документацию по API PySpark для последних поддерживаемых API.

Сведения о других ограничениях см. в разделе об ограничениях бессерверных вычислений.

Загрузка больших разностных таблиц с помощью томов

Для больших таблиц Delta, которые слишком большие для преобразования toPandas(), экспортируйте данные в том каталога Unity и загружайте их непосредственно с помощью PyTorch или 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")

Этот подход позволяет избежать накладных расходов Spark во время обучения и хорошо подходит как для рабочих процессов обучения с одним GPU, так и для распределенных рабочих процессов.

Загрузка неструктурированных данных из томов

Для неструктурированных данных, таких как изображения, звуковые и текстовые файлы, используйте тома каталога Unity. В следующем примере показано, как считывать файлы с носителя и использовать их в PyTorch Dataset:

# Read files from a UC volume
volume_path = "/Volumes/catalog/schema/my_volume/images/"

from torch.utils.data import Dataset
import os
from PIL import Image

class ImageDataset(Dataset):
    def __init__(self, root_dir):
        self.file_list = [os.path.join(root_dir, f) for f in os.listdir(root_dir)]

    def __len__(self):
        return len(self.file_list)

    def __getitem__(self, idx):
        img = Image.open(self.file_list[idx])
        return img

Загрузка данных внутри @distributed декоратора

При использовании безсерверного API GPU для распределенного обучения переместите код загрузки данных в декоратор @distributed. Размер набора данных может превышать максимальный размер, разрешенный пикл, поэтому рекомендуется генерировать набор данных внутри декоратора, как показано ниже:

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

Производительность загрузки данных

/Workspace и /Volumes размещаются в удаленном хранилище каталога Unity. Если набор данных хранится в каталоге Unity, скорость загрузки данных ограничена доступной пропускной способностью сети. Если вы обучаете несколько эпох, рекомендуется сначала скопировать данные локально, а именно в директорию /tmp, которая размещается на быстром хранилище SSD NVMe.

Если набор данных велик, следующие методы могут повысить производительность:

• Сохраняйте данные в кэш локально для многократного обучения. Скопируйте наборы данных в /tmp для более быстрого доступа в разных эпохах.

  import shutil
  shutil.copytree("/Volumes/catalog/schema/volume/dataset", "/tmp/dataset")
  

• Параллелизация получения данных. Используйте torch DataLoader с несколькими рабочими процессами для параллельного выполнения загрузки данных и вычислений на GPU. Установите num_workers значение не менее 2. Чтобы повысить производительность, увеличьте num_workers (что увеличивает параллельную обработку чтения) или prefetch_factor (что увеличивает количество элементов, предварительно выбираемых каждым рабочим):

  from torch.utils.data import DataLoader
  
  loader = DataLoader(
      dataset,
      batch_size=32,
      num_workers=2,
      prefetch_factor=2,
      pin_memory=True
  )
  

• Используйте Spark Connect для больших табличных наборов данных. Spark Connect поддерживает большинство API PySpark и обрабатывает распределенные операции чтения эффективно.

Потоковая передача наборов данных

Для очень больших наборов данных, которые не соответствуют памяти, используйте подходы потоковой передачи:

• PyTorch IterableDataset для пользовательской логики потоковой передачи.
• Наборы данных Hugging Face с потоковой передачей для наборов данных, размещенных на хабе или в томах.
• Ray Data для распределенной пакетной обработки данных.