Yapay zeka çalışma zamanına veri yükleme

Önemli

Bu özellik Genel Önizleme aşamasındadır.

Veri ve model varlıkları, büyük dil modelleri (LLM'ler) ve vizyon-dil modelleri (VLM'ler) için derin öğrenme ve eğitim sonrası iş yüklerinde kritik öneme sahiptir. AI Runtime ile tüm veri ve model varlıklarına Unity Kataloğu üzerinden erişilir:

• Unity Kataloğu ciltleri: öncelikle büyük veri setleri ve yapılandırılmamış dosyalar için, görseller, ses ve metin dahil olmak üzere kullanılır.
• Unity Kataloğu tabloları: yapılandırılmış ve tablo veriler için kullanılır, Spark Connect üzerinden erişilir.

Ciltleriniz ve tablolarınız Unity Kataloğu'nda kayıtlı olmalı ve kullanıcı veya hizmet yöneticisiniz tarafından erişilebilir olmalıdır.

Yapılandırılmamış veriler için Unity Catalog birimi

Unity Kataloğu ciltleri, yapılandırılmış, yarı yapılandırılmış ve yapılandırılmamış veriler dahil olmak üzere herhangi bir formatta tablo dışı verilere yönetimli erişim sağlar. AI Çalışma Zamanında, hacimler büyük veri setlerine, metne, model varlıklarına ve model kontrol noktalarına erişim için birincil mekanizmadır.

Kullanıcılar, yerel diskteki dosyalarla çalışmaya benzer şekilde tanıdık dosya sistemi işlemlerini kullanarak Unity Kataloğu hacimlerinde dosyaları listeleyebilir, okuyabilir ve yazabilirler:

import os

dir_path = "/Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir"
file_path = os.path.join(dir_path, "test_file")

os.makedirs(dir_path, exist_ok=True)

# Write to the file
with open(file_path, "w") as file:
    file.write("Hello, World!")

Benzer şekilde, shell işlemleri de aynı şekilde çalışır:

%sh ls -l /Volumes/<catalog-name>/<schema-name>/<volume-name>
%sh mkdir -p /Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir
%sh touch /Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/test_file

Unity Kataloğu hacimlerinin birkaç özelliği, onları makine öğrenimi iş yükleri için oldukça uygun kılmaktadır:

• Dağıtık depolama: Unity Catalog, dağıtık depolama ile desteklenir; bu da AI Çalışma Zamanı iş yüklerinin platform genelinde hem dizüstü bilgisayarlardan hem de CLI tabanlı iş yüklerinden veri okuyup yazmasına ve varlıklarını modellemesine olanak tanır.
• ML erişim kalıpları için optimize edilmiş: Temel depolama ve erişim yolları, özellikle sıralı okuma ve yazma işlemlerine sahip büyük ML iş yükleri için optimize edilmiştir. Bu da Unity Catalog’u eğitim verilerinin yüklenmesi, model varlıklarının yüklenmesi ve model kontrol noktalarının kaydedilmesi için çok uygun kılar.
• Dosya sistemi benzeri erişim: Kullanıcılar, yerel diskteki dosyalarla çalışmaya benzer şekilde tanıdık dosya sistemi işlemleriyle Unity Kataloğu hacimlerinde dosyaları listeleyebilir, okuyabilir ve yazabilir.

Otomatik arka plan commitleri sayesinde kullanıcılar Unity Kataloğu hacim verilerine tutarlı erişim bekleyebilirler:

• Yazıyor: AI Çalışma Zamanı, otomatik olarak yazma işlemlerini yapar ve değişiklikleri aynı Unity Kataloğu hacmine erişen diğer uygulamalar ve iş yüklerine görünür hale getirir.
• Okuyor: AI Runtime, kullanıcıdan herhangi bir açık yenileme veya senkronizasyon işlemi gerektirmeden hacimdeki değişiklikleri otomatik olarak algılar.

Ses düzeyi performansını ayarlayın

Daha önce belirtildiği gibi, Unity Kataloğu hacimleri dağıtık depolama ile desteklenir ve ardışık okuma ve yazma işlemleriyle büyük dosyalar için optimize edilmiştir.

AI Runtime'dan en iyi performansı elde etmenize yardımcı olabilecek birkaç ipucu:

• Veriyi daha büyük dosyalara birleştirin: Mümkünse, verileri daha az, daha büyük dosyalara birleştirin; dosya başına yaklaşık 1 GiB ile 10 GiB arasında. Bu, AI Runtime'ın verileri agresif bir şekilde ön getirmesini ve otomatik olarak neredeyse optimal ardışık okuma performansına ulaşmasını sağlar.

• Küçük dosyalı iş yükleri için yerel disk kullanın: Eğer iş yükünüz çok sayıda küçük dosya içeriyorsa, dosyaları işlemeden önce paralel kopyalar kullanarak yerel diske kopyalamayı düşünün. Bu, hacim üzerinden birçok küçük dosyaya tekrar tekrar erişmenin getirdiği yükü azaltabilir.

  # Recommended using parallel copy (256 concurrency in this example, you can tune)
  #
  # This takes only 22 seconds to copy 15,375 150KiB small image files.
  %sh cd /Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/ && find . -type f -print0 | xargs -0 -P 256 -I {} cp --parents "{}" /tmp/
  
  
  # !!! Avoid doing this !!!
  #
  # Because the files are copied in serial, this copies the same 15,375 150KiB small image files much more slowly.
  # %sh cp -r /Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/* /tmp
  

• Makine öğrenimi iş yükleriniz için kullanabilirsiniz UCVolumeDataset . Yukarıda açıklanan optimizasyonları içererek Unity Kataloğu hacimlerinden verimli veri erişimi ve yükleme sağlamaktadır. Diğer ayrıntılar için aşağıdaki bölümlere bakın.

Yapılandırılmamış verileri UCVolumeDataset ile yükleyin

Unity Kataloğu birimlerinde depolanan görsel, ses ve metin dosyaları gibi yapılandırılmamış veriler için UCVolumeDataset modülünden databricks.air.data kullanabilirsiniz. UCVolumeDataset , birimdeki her dosyayı ilk erişimde hızlı bir yerel önbelleğe kopyalayan ve önbelleğe alınmış yerel dosya yolunu veren bir PyTorch'tır IterableDataset . Aksi takdirde el ile uygulayacağınız performans ve dağıtım sorunlarını ele alır:

• Yerel önbellekleme Dosyalar, ilk erişimde FUSE bağlama noktasından yerel bir önbellek dizinine kopyalanır ve sonraki erişimlerde önbellekten sunulur; bu sayede birden çok epoch süren eğitim sırasında birim yeniden okunmaz.
• Otomatik bölümleme. torch.distributed başlatıldığında, dosyalar rank'ler arasında bölümlenir ve ardından DataLoader çalışanı arasında daha da bölünür; böylece her (rank, worker) çifti ek yapılandırma gerekmeden çakışmayan bir dilim alır.

Uyarı

UCVolumeDataset ve databricks.air.data.DataLoader, databricks-sdk-air paketinden gelir. data ekstra ile kurun; bu, uyumlu bir torch ögesini de çeker:

%pip install "databricks-sdk-air[data]"

UCVolumeDataset ham yerel dosya yollarını verir. Bu dosyaları tensörlere çözümlemek için, yol akışını tüketen ve ayrıştırma mantığınızı uygulayan ikinci bir IterableDataset ile sarın. Bu, G/Ç ve ayrıştırma sorunlarını ayrı tutar.

from databricks.air.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)

Sarmalayıcı zaten önbelleğe alınmış yerel yolları alır, bu nedenle ayrıştırma adımı birime hiç erişmez. Ekleme, belirteç oluşturma veya filtreleme için ek sarmalayıcıları zincirleyebilirsiniz.

En iyi performans için, varsayılan PyTorch UCVolumeDataset yerine databricks.air.data.DataLoader ile DataLoader'i eşleştirin. AI Runtime I/O için ayarlanmış ve GPU hesaplar yaparken dosyaları eşzamanlı olarak alıyor ve önbellekliyor.

Hacimlerde kontrol noktası modelleri

Modelinizi kontrol noktası olarak en son anlık görüntüden eğitime devam edebilmek veya bir çöküşten sonra toparlanmak için, Unity Catalog hacimlerini yerel dosya sistemi gibi kullanabilirsiniz.

Databricks, hem tek GPU hem de çoklu GPU iş yüklerinde daha iyi performans için dağıtık kontrol noktası (DCP) kullanılmasını önerir. Databricks mühendislik blogundan Hızlı ve hata toleranslı PyTorch AI Runtime eğitimine bakınız.

import torch.distributed.checkpoint as dcp
from torch.distributed.checkpoint.state_dict import get_state_dict, set_state_dict

import databricks.air.data

checkpoint_path = "/Volumes/my-catalog/my-schema/my-volume/checkpoints/step_1000"

# Save
model_sd, optim_sd = get_state_dict(model, optimizer)
state_dict = {"model": model_sd, "optim": optim_sd, "step": 1000}
dcp.async_save(
    state_dict,
    storage_writer=databricks.air.data.UCVolumeWriter(checkpoint_path))

# Load
model_sd, optim_sd = get_state_dict(model, optimizer)
state_dict = {"model": model_sd, "optim": optim_sd}
dcp.load(
    state_dict,
    storage_reader=databricks.air.data.UCVolumeReader(checkpoint_path))

set_state_dict(
    model,
    optimizer,
    model_state_dict=state_dict["model"],
    optim_state_dict=state_dict["optim"],
)

Monolitik torch.save yaklaşımı da işe yarıyor.

• Tek GPU’lu model için denetim noktası oluşturma,

  # The monolithic torch.save approach for single GPU chip
  
  # Save
  torch.save({"model": model.state_dict(), "opt": optimizer.state_dict()},
             "/Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/ckpt.pt")
  
  
  # Load
  ckpt = torch.load(
      "/Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/ckpt.pt",
      weights_only=True)
  model.load_state_dict(ckpt["model"])
  optimizer.load_state_dict(ckpt["opt"])
  

• Torchrun ile başlatılan dağıtık eğitim için,

  # The monolithic torch.save approach for multi-GPU distributed training.
  # This snippet assumes your launcher has already called
  # dist.init_process_group(...).
  
  import os
  import torch.distributed as dist
  
  # Save only on rank 0.
  if dist.get_rank() == 0:
      torch.save({"model": model.state_dict(), "opt": optimizer.state_dict()},
                 "/Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/ckpt.pt")
  
  # Wait for rank 0 to finish writing before any rank reads.
  dist.barrier()
  
  # Load on ALL ranks (map to current rank's local GPU).
  local_rank = int(os.environ["LOCAL_RANK"])
  ckpt = torch.load(
      "/Volumes/<catalog-name>/<schema-name>/<volume-name>/sub-dir/ckpt.pt",
      map_location=f"cuda:{local_rank}",
      weights_only=True)
  model.load_state_dict(ckpt["model"])
  optimizer.load_state_dict(ckpt["opt"])
  

Tablo verilerini yükleme

Delta tablolarından tablosal makine öğrenmesi verilerini yüklemek için Spark Connect'i kullanın.

Tek düğümlü eğitim için PySpark yönteminitoPandas() kullanarak Apache Spark DataFrames'i pandas DataFrames'e dönüştürebilir ve ardından isteğe bağlı olarak PySpark yönteminito_numpy() kullanarak NumPy biçimine dönüştürebilirsiniz.

Uyarı

Spark Connect, analiz ve ad çözümlemeyi yürütme zamanına erteler, bu da kodunuzun davranışını değiştirebilir. Bkz. Spark Connect ile Spark Klasik'i karşılaştırma.

Spark Connect, Spark SQL, Spark üzerinde Pandas API'si, Yapılandırılmış Akış ve MLlib (DataFrame tabanlı) dahil olmak üzere çoğu PySpark API'sini destekler. Desteklenen en son API'ler için PySpark API başvuru belgelerine bakın.

Diğer sınırlamalar için bkz. Sunucusuz işlem sınırlamaları.

Unity Kataloğu hacimleri kullanarak büyük Delta tabloları yükleyin

ile toPandas()dönüştürülemeyecek kadar büyük olan büyük Delta tabloları için verileri bir Unity Kataloğu birimine aktarın ve PyTorch veya Hugging Face kullanarak doğrudan yükleyin:

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

Bu yaklaşım, eğitim sırasında Spark ek yükünü önler ve hem tek GPU hem de dağıtılmış eğitim iş akışları için iyi çalışır.