Python istemcisiyle bulut depolama için Distributed Accelerator kullanın (önizleme)

Bulut depolama için dağıtık hızlandırıcı (DACS), dağıtık önbellek üzerinden Azure Blob verilerini okumak için yerel bir Python istemcisi sağlar.

Bu makaleyi kullanmadan önce AKS'te bulut depolama için dağıtık hızlandırıcı ile çalışmaya başlama makalesini tamamlayın.

Yerel Python istemcisini kur

PyPI'den paketi iş yükü imajınıza yükleyin:

python3 -m pip install "dacs-client==0.1.0"

Mevcut bir iş yükünü güncelliyorsanız, paket pinini imaj derleme tanımında güncelleyin, görüntüyü yeniden oluşturup itin ve güncellenmiş imajla iş yükü podlarını yeniden dağıtın. Paketi yerel olarak yüklemek, çalışan konteynerleri güncellemez. Görüntü oluşturma ve push komutları için, İstemciyi iş yükü görüntüsüne ekle konusuna bakın.

Paket, Python 3.8 veya daha yenisini ve glibc 2.35 veya daha yenisine sahip bir Linux ortamını gerektirir. manylinux_2_35 x86-64 ve arm64 için yerel kütüphane ve paylaşılan kütüphane bağımlılıkları dahil, bağımsız wheel paketleri sağlar. Bağımlılık azure-identity otomatik olarak kurulur.

Hızlı başlangıç, x86-64 düğümleriyle python:3.12-slim-bookworm üzerinde Python 3.12 kullanır.

İş yükü konteyneri içindeki kurulumu doğrulayın:

python3 -c "import importlib.metadata; import py_dacs_client; print(importlib.metadata.version('dacs-client'))"

Komut 0.1.0 yazdırmalıdır. Dağıtım adı dacs-client, ancak Python içe aktarma adı py_dacs_client.

İstemciyi yapılandırma

Bulut depolama için dağıtık hızlandırıcı kullanan her podda her iki önbellek keşif değişkenini ayarlayın:

env:
  - name: CACHE_DISCOVERY_URL
    value: <discovery-service-hostname>
  - name: CACHE_SERVER_PORT
    value: "9065"

Important

Her iki değişken de gereklidir. Eğer herhangi biri eksikse, istemci Cache not configured. Working in Remote Blob mode only. günlüğe kaydeder ve Azure Blob Depolama'a doğrudan erişir. İşlem, önbellek kullanılmadan da başarılı olabilir.

Önbellek kaynağınız ve ad alanınız için keşif hizmeti ana bilgisayar adını kullanın, örneğin, cache-sample-discovery.<namespace>.svc.cluster.local, bağlantı noktası 9065 ile.

İstemciyi oluşturmadan önce Blob kimlik doğrulamasını yapılandırın. Aşağıdaki örnek, federated AKS iş yükü kimliğini değil, düğümün Instance Metadata Service (IMDS) üzerinden yönetilen AKS kubelet kimliğini kullanır. AZURE_CLIENT_ID değerini kubelet kimliğinin istemci kimliğine ayarlayın:

env:
  - name: AZURE_CLIENT_ID
    value: <kubelet-identity-client-id>
  - name: MSI_ENDPOINT
    value: "http://169.254.169.254/metadata/identity/oauth2/token"
  - name: MSI_SECRET
    value: "required-nonempty-value"

Pod IMDS'ye ulaşabilmeli. Bu yerel istemci yapılandırması için, MSI_SECRET gerçek bir sır veya kimlik bilgisi değil, zorunlu olan boş olmayan bir uyumluluk belirtecidir. O değere bir kimlik bilgisi koymayın. Bu ayarlar bu entegrasyona özgüdür; bunları Run:ai kimlik doğrulama yapılandırmasına kopyalamayın.

Salt okunur çıkarım için, kubelet kimliğine model container'ı kapsamında Storage Blob Data Reader rolünü verin. Bkz. Model verilerine okuma erişimi verme. Storage Blob Data Contributor yalnızca uygulamanın veri yazması gerektiğinde kullanın. Çalışma zamanı okumaları için ayrı bir yükleyici yazma iznine gerek yoktur.

Bu kanonik yol formatını kullanın:

az://<account>.blob.core.windows.net/<container>/<path>

Yerel katman, bir süreçte ilk istemci oluşturulduğunda yapılandırmayı okur. O istemciyi oluşturmadan önce ortamı ve istemci ayarlarını yapılandırın. Daha sonraki istemciler zaten geçerli olan yerel yapılandırmayı değiştirmez.

Önbellek ıskalarını ve eşzamanlı okumaları anlayın

Birden fazla istemci aynı zamanda aynı eksik veriyi talep ettiğinde, her istemci verileri bağımsız olarak Azure Blob Depolama'tan alabilir. Koordineli soğuk önbellek popülasyonu bu önizlemede mevcut değil.

Önbelleğin doldurulması eşzamansızdır. Bir okuma, verileri önbellekte kullanılabilir olmadan önce tamamlanabilir, bu yüzden iki anlık okuma, geri dönen baytlar eşleşse bile ıskalayabilir.

Büyük senkronize ölçekli olaylardan önce, seçilen model bloblarını çıkarım replikalarını başlatmadan önce DACS üzerinden ısıtın. Özel bir ısınma podu, okuma komutları ve dağıtım doğrulaması için dağıtımdan önce Prefetch model data before deployment belgesini izleyin.

Başarılı bir okuma Azure Blob Depolama geri dönüşünden gelebilir. Okumaların DACS kullandığını doğrulamak için:

  1. Okumadan önce, Cache kaynağınız için tüm önbellek sunucusu podları genelinde toplanmış cache_server_request_counter{request_type="Download",status="Success"} değerini kaydedin. Etiket değerleri büyük/küçük harfe duyarlıdır.
  2. Okumalar arasında gecikmeyle aynı blob aralığını tekrar tekrar okuyun. Hızlı başlatma, bir saniye aralıklı beş okuma kullanır; bu dizi, doldurma işleminin tamamlandığını garanti etmez. Sunucudaki başarılı indirme sayısının arttığını ve istemcinin önbellek isabetlerini bildirdiğini doğrulayın. Veri bütünlüğünü doğrulamak için dönen bayt sayılarını ve sağlama toplamlarını kontrol edin; tek başına daha hızlı süreler, verilerin önbelleğe alındığını kanıtlamaz.
  3. Önbellek trafiğini kaynak okumalarından ve geri dönüşlerden ayırt etmek için istemci ve sunucu metriklerini Azure Blob telemetrisiyle ilişkilendirin.

İstemci metriklerini ve sunucu sayaçlarını toplama komutları için DACS istek yolunu doğrulama bölümüne bakın.

Yerel istemciyi kullanın

Yerel istemci, asenkron yöntemler ve bunların senkron eşdeğerlerini sunar. Veriyi önceden tahsis edilmiş yerel tamponlara okur. Aşağıdaki örneği çalıştırmadan önce, MODEL_PATH değerini mevcut bir model parçasının kurallı az:// yolu olarak ayarlayın; bu örnek ilk 4 MiB’ye kadarını okur:

import asyncio
import os

from py_dacs_client.dacs_client import DACSCacheClient
from py_dacs_client.dacs_utils import BoundedBufferManager


async def main():
    path = os.environ["MODEL_PATH"]
    client = DACSCacheClient()
    buffers = BoundedBufferManager(max_memory_limit=64 * 1024 * 1024)

    try:
        size = min(await client.get_file_size(path), 4 * 1024 * 1024)
        if size == 0:
            print("read 0 bytes")
            return

        buffer = await buffers.acquire_buffer(size)
        try:
            bytes_read = await client.read_blob(path, buffer, size, 0)
            print(f"read {bytes_read} bytes")
        finally:
            await buffers.release_buffer(buffer)
    finally:
        await client.shutdown()


asyncio.run(main())

Yaygın okuma yöntemleri şunlardır:

Asenkron yöntem Senkron yöntem Description
get_file_size(path) get_file_size_sync(path) Blob boyutunu bayt cinsinden alın.
get_file_stat(path) get_file_stat_sync(path) Model dosyası meta verilerini alın.
read_blob(path, buffer, size, offset) read_blob_sync(...) Bir bayt aralığını önceden ayrılmış bir arabelleğe okuyun.
list_dir(path) list_dir_sync(path) Dizin girişlerini listeleyin.
get_file_handle(path, flags) get_file_handle_sync(...) Yeniden kullanılabilir bir tanıtıcı açın.
release_file_handle(handle) release_file_handle_sync(handle) Yeniden kullanılabilir bir tanıtıcıyı kapatın.
shutdown() shutdown_sync() Bağlantıları serbest bırakın ve son metriklerin çıktısını verin.

Asenkron list_dir(path) yöntemi bir yineleyici döndürür; dizin girişlerini tüketmek için async for kullanın.

Her bayta ihtiyaç duymayan model formatları için aralık okumaları kullanın. Bulut depolama için dağıtık hızlandırıcı yalnızca istenen aralıkları kapsayan parçaları saklar.

Okuma performansını optimize et

Uygulama okuma eşzamanlılığı ve istenen bayt aralığı boyutunu birlikte ayarlayın. En iyi ayarlar, erişim deseninize, pod CPU ve bellek sınırlarınıza, istemci ve önbellek düğümü ağ bant genişliğine ve okumaların önbellekten karşılanıp karşılanmadığına veya Blob Depolama'a geri dönüp dönmediğine bağlıdır.

Bu bölümdeki değerleri iş yükü kıyaslamaları için başlangıç noktası olarak kullanın, garanti edilen performans önerileri olarak değil. Değişiklikleri dağıtmadan önce etkilerini kendi iş yükünüzle ölçün.

Eşzamanlılığı ayarla

Eşzamanlılık, uygulamanızın devam eden okuma sayısıdır. Her read_blob çağrıyı bekleyip bir sonraki çağrıyı başlatan bir döngü, asenkron API kullansa bile ardışıktır.

  • Bağımsız bayt aralığı veya blob okumalarını eşzamanlı olarak gerçekleştirin. Aktif okuma sayısını, asyncio.Semaphore veya sabit sayıda çalışan göreviyle sınırlandırın. Büyük bir veri setinin her aralığı için aynı anda bir görev oluşturmayın.
  • Dört eşzamanlı okuma gibi küçük bir sınırla başlayın ve bunu sekiz ve 16 ile karşılaştırın. Yalnızca çıktı hızı arttıkça eşzamanlılığı artırın, kabul edilemez gecikme, bellek kullanımı, hata veya kısıtlama olmadan.
  • Yerel istemci, concurrent.futures.Executor yapıcı parametresi aracılığıyla executor kabul eder. Python iş parçacığı havuzunu açıkça kontrol etmek için bir DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)) geçirin, örneğin ThreadPoolExecutor. Yürütücüye referans verin, tüm okumalar bitene kadar canlı tutun ve await client.shutdown() çağrıldıktan sonra kapatın.
  • Daha büyük bir iş parçacığı havuzu, ardışık çağrıları eşzamanlı hâle getirmez. Uygulama görevlerini ve executor çalışanlarını bağlayın ve her aktif okuma için ayrı bir tampon kullanın.

Tüm çalışan süreçleri ve podlar arasında eşzamanlılığı hesaba katın. Ölçeklendirme sırasında her podun eşzamanlılığını artırmak, aynı önbellek düğümlerini aşırı yükleyebilir veya soğuk önbellek kaçırmalarında tekrarlanan Blob indirmelerini artırabilir.

Önbellek parçası boyutunu yapılandır

Cache-chunk boyutunu yapılandırmak için uygulama podlarında SI_streamingChunkSize öğesini bayt olarak ayarlayın. Örneğin, aşağıdaki ayar 3-MiB parçalar kullanır:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Bu değişkeni bir süreçteki ilk istemciyi oluşturmadan önce ayarlayın. Bunu değiştirdikten sonra uygulama podlarını yeniden başlatın.

Varsayılan önbellek parçası boyutu 32 MiB'dir. read_blob(path, buffer, size, offset) için size bağımsız değişkeni, önbellek-parça boyutunu değil, uygulamanın istediği bayt aralığını belirtir.

Erişim düzeni Uygulama okuma kılavuzu
Büyük ardışık model okumaları Daha büyük önbellek parçaları, istenen aralık boyutları ve sınırlı eşzamanlılığı karşılaştırın. Bir kerede bir ayarı değiştirin.
Küçük veya rastgele okuma işlemleri Sadece uygulamanın ihtiyaç duyduğu baytları talep edin. İstenen aralığın dışından alınan verileri azaltmak için daha küçük önbellek parçalarını karşılaştırın.
Kapasiteyle sınırlı önbellekler Önbellek doluluğunu ve önbellekten çıkarma davranışını, yalnızca uygulama okuma arabelleği boyutlarından yola çıkarak tahmin etmek yerine ölçün.

Bütçe uygulaması okuma tamponlarını önbellek parçalarından ayrı olarak okur. Örneğin, sekiz eşzamanlı 4-MiB read_blob istek, yapılandırılmış önbellek parçası boyutuna bakılmaksızın 32 MiB uygulama okuma tamponu kapasitesi gerektirir. Yerel istemci tahsisleri, sıraya alınan çalışmalar, Python nesneleri ve model işleme için ek pod belleği ayırın. Arabellekleri yalnızca eşzamanlılık sınırına ulaştıktan sonra alın ve her arabelleği finally bloğunda serbest bırakın.

Warning

BoundedBufferManager(max_memory_limit=...) uygulama okuma tamponu kapasitesini sınırlar; toplam istemci belleğini veya yerel yükleme arabelleğe almasını sınırlamaz. 0.1.0 içinde, yerel istemcinin varsayılan asenkron yükleme-tampon bütçesi 400 GiB'dir. Bu bütçe bir ön tahsis ya da güvenli bir Pod-bellek sınırı değildir. Büyük soğuk önbellek okumaları daha küçük bir Pod'u tüketebilir. Bellek kısıtlı iş yükleri için, senkron doldurmayı (SI_cacheEnableSyncUpload=true) veya sınırlı asenkron yükleme bütçesini (SI_cacheMaxUploadBufferSizePerProcessGB) test edin. İlk istemciyi oluşturmadan önce bu değerleri ayarlayın ve uygulama okuma tamponları ve model işleme süreçlerinden ayrı olarak yerel arabellekleri bütçeleyin.

Sınırları artırmadan önce ölçün

  1. Sabit bir veri seti, pod sayısı ve kaynak sınırlarıyla bir temel oluşturun. Aktarım hızını, okuma gecikmesini, CPU, bellek ve ağ kullanımını ölçün.
  2. Cold-cache ve warm-cache okumaları için benchmark işlemlerini ayrı ayrı yapın. Önbellek trafiğini doğrudan Blob'a geri dönüşten ayırt etmek için cache_server_request_counter ve Azure Blob telemetrisini inceleyin.
  3. Her seferinde bir ayarı değiştirin: eşzamanlılık, uygulama okuma boyutu veya önbellek parçası boyutu. Sonuçları karşılaştırırken veri seti ve diğer ayarları tutarlı tutun.
  4. Verim sabitlendiğinde veya gecikme, kaynak baskısı, hatalar ya da Blob throttling arttığında bir ayarı artırmayı bırakın.

Önbellek tarafı gözlemleri için bkz. Bulut depolama için dağıtılmış hızlandırıcıyı izleme.

Hatalarla başa çıkma

Tüm DACS istisnaları DACSErrorşundan türemiştir:

from py_dacs_client import DACSError, DACSNotFoundError

try:
    load_model_data()
except DACSNotFoundError:
    handle_missing_model()
except DACSError as error:
    handle_cache_or_storage_error(error)
Exception Meaning
DACSNotFoundError Blob yolu veya dizini mevcut değil.
DACSExistsError Bir uygulama, üzerine yazma etkinleştirilmeden mevcut bir bloba yazmaya çalışır.
DACSAuthenticationError Kimlik bilgileri eksik veya geçersiz.
DACSPermissionError Kimlik gerekli Blob rolünden yoksun.
DACSTimeoutError Bir önbellek veya Blob işleminin süresi doldu.
DACSNetworkError Önbellek sunucusu veya Blob uç noktasına ulaşılamıyor.
DACSInvalidInputError Bir yol ya da bağımsız değişken hatalı biçimlendirilmiş.
OperationNotSupportedError İstenen API henüz geliştirilmedi.

Günlüğe kaydetmeyi denetleme

İstemcinin Python ve yerel katmanlar için ayrı kayıt kontrolleri vardır. Her iki değişkeni de her iki katmandan gelen ayrıntılı çıktıyı bastıracak şekilde ayarlayın:

env:
  - name: DACS_DATA_LOG_LEVEL
    value: WARNING
  - name: STORAGE_LOG_LEVEL
    value: WARNING

DACS_DATA_LOG_LEVELPython loglamasını kontrol eder. STORAGE_LOG_LEVEL yerel istemci günlüğe kaydetmeyi kontrol eder.

Sonraki Adımlar

  • Son güncelleştirme: