Verwenden Sie den verteilten Beschleunigerdienst für Cloud-Speicher mit dem Python-Client (Vorschau)

Der Distributed Accelerator for Cloud Storage (DACS) bietet einen nativen Python-Client zum Lesen von Azure Blob-Daten über den verteilten Cache.

Schließen Sie Erste Schritte mit dem verteilten Beschleuniger für Cloudspeicher auf AKS ab, bevor Sie diesen Artikel lesen.

Installiere den nativen Python-Client

Installieren Sie das Paket aus PyPI in Ihrem Workload-Image:

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

Wenn du eine bestehende Workload aktualisierst, aktualisiere den Package-Pin in der Image-Build-Definition, baue das Image neu auf und pushe es und deploye die Workload-Pods mit dem aktualisierten Image neu. Die lokale Installation des Pakets aktualisiert die laufenden Container nicht. Für Image-Build- und Push-Befehle siehe den Client zu einem Workload-Image hinzufügen.

Das Paket benötigt Python 3.8 oder neuer sowie eine Linux-Umgebung mit glibc 2.35 oder neuer. Es stellt eigenständige manylinux_2_35 Wheels für x86-64 und arm64 bereit, einschließlich der nativen Bibliothek und deren Shared-Library-Abhängigkeiten. Die Abhängigkeit azure-identity wird automatisch installiert.

Der Quickstart verwendet Python 3.12 auf python:3.12-slim-bookworm mit x86-64-Knoten.

Überprüfen Sie die Installation im Workload-Container:

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

Der Befehl muss 0.1.0 ausgeben. Der Vertriebsname ist dacs-client, aber der Python-Importname ist py_dacs_client.

Konfigurieren des Clients

Setzen Sie beide Cache-Discovery-Variablen für jeden Pod, der einen verteilten Beschleuniger für Cloud-Speicher verwendet:

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

Important

Beide Variablen sind erforderlich. Wenn eines davon fehlt, protokolliert der Client Cache not configured. Working in Remote Blob mode only. und greift direkt auf Azure Blob Storage zu. Die Operation kann auch ohne Nutzung des Caches erfolgreich sein.

Verwenden Sie den Hostnamen des Discovery-Dienstes für Ihre Cache-Ressource und Ihren Namensraum, zum Beispiel cache-sample-discovery.<namespace>.svc.cluster.local, mit Port 9065.

Konfigurieren Sie die Blob-Authentifizierung, bevor Sie den Client erstellen. Das folgende Beispiel verwendet die verwaltete AKS-Kubelet-Identität über den Instance Metadata Service (IMDS) des Knotens, nicht die föderierte AKS-Workload-Identität. Setzen Sie AZURE_CLIENT_ID auf die Client-ID der Kubelet-Identität:

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"

Der Pod muss IMDS erreichen können. Für diese Native-Client-Konfiguration ist MSI_SECRET ein erforderlicher, nicht leerer Kompatibilitätsmarker, kein echtes Geheimnis und keine echte Anmeldeinformation. Setzen Sie in diesen Wert keine Anmeldeinformation ein. Diese Einstellungen sind spezifisch für diese Integration; kopiere sie nicht in die Run:ai-Authentifizierungskonfiguration.

Für die Read-Only-Inferenz gewähren Sie der Kubelet-Identität die Rolle Storage Blob Data Reader mit Geltungsbereich für den Modellcontainer. Siehe Lesezugriff auf Modelldaten gewähren. Verwenden Sie Storage Blob Data Contributor nur, wenn die Anwendung Daten schreiben muss. Eine separate Schreibberechtigung des Uploaders ist für Laufzeit-Reads nicht erforderlich.

Verwenden Sie dieses kanonische Pfadformat:

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

Die native Schicht liest die Konfiguration, wenn der erste Client in einem Prozess erstellt wird. Konfigurieren Sie die Umgebung und die Client-Einstellungen, bevor Sie diesen Client erstellen. Spätere Clients ändern die bereits gültige native Konfiguration nicht.

Verstehen Sie Cache-Misses und gleichzeitige Lesezugriffe

Wenn mehrere Clients gleichzeitig dieselben fehlenden Daten anfordern, kann jeder Client die Daten unabhängig aus Azure Blob Storage abrufen. Eine koordinierte Cold-Cache-Population ist in dieser Vorschau nicht verfügbar.

Die Cache-Population ist asynchron. Ein Lesevorgang kann abgeschlossen werden, bevor seine Daten aus dem Cache verfügbar sind, sodass zwei sofortige Lesevorgänge beide fehlschlagen können, selbst wenn die zurückgegebenen Bytes übereinstimmen.

Erwärmen Sie vor großen synchronisierten Skalierungsereignissen die ausgewählten Modellblobs durch DACS, bevor Sie mit den Inferenzreplikaten beginnen. Befolgen Sie Prefetch-Modelldaten vor der Bereitstellung für einen dedizierten Warm-up-Pod, Leseanweisungen und die Rollout-Verifikation.

Ein erfolgreicher Lesevorgang kann durch den Azure Blob Storage-Fallback erfolgen. So überprüfen Sie, ob Lesezugriffe DACS verwenden:

1. Vor dem Lesen erfassen Sie cache_server_request_counter{request_type="Download",status="Success"}, summiert über alle Cache-Server-Pods für Ihre Cache-Ressource. Werte von Labels sind groß-/kleinschreibungssensitiv.
2. Lesen Sie denselben Blob-Bereich wiederholt mit einer Verzögerung zwischen den Lesungen. Der Quickstart verwendet fünf Reads, die eine Sekunde auseinanderliegen; diese Sequenz garantiert nicht, dass die Befüllung abgeschlossen ist. Bestätigen Sie, dass die Anzahl erfolgreicher Server-Downloads steigt und der Client Cache-Hits meldet. Überprüfe die zurückgegebenen Byte-Zahlen und Prüfsummen auf die Datenintegrität; schnelleres Timing allein ist kein Beweis für Caching.
3. Korrelieren Sie Client- und Servermetriken mit Azure Blob-Telemetrie, um Cache-Verkehr von Origin-Reads und Fallback zu unterscheiden.

Für Befehle zur Erfassung von Client-Metriken und Serverzählern siehe Verify the DACS request path.

Verwende den nativen Client

Der native Client stellt asynchrone Methoden und synchrone Gegenstücke bereit. Es liest in vorab zugewiesene native Puffer ein. Setzen Sie MODEL_PATH auf den kanonischen az:// Pfad eines bestehenden Model-Shards, bevor Sie das folgende Beispiel ausführen, das bis zu den ersten 4 MiB liest:

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

Gängige Lesemethoden sind:

Asynchrone Methode	Synchrone Methode	Description
get_file_size(path)	get_file_size_sync(path)	Ruft die Größe des Blobs in Byte ab.
get_file_stat(path)	get_file_stat_sync(path)	Metadaten der Modelldatei abrufen.
read_blob(path, buffer, size, offset)	read_blob_sync(...)	Lesen Sie einen Bytebereich in einen vorab zugewiesenen Puffer.
list_dir(path)	list_dir_sync(path)	Verzeichniseinträge auflisten.
get_file_handle(path, flags)	get_file_handle_sync(...)	Öffne ein wiederverwendbares Handle.
release_file_handle(handle)	release_file_handle_sync(handle)	Schließe ein wiederverwendbares Handle.
shutdown()	shutdown_sync()	Verbindungen freigeben und endgültige Kennzahlen ausgeben.

Die asynchrone list_dir(path) Methode liefert einen Iterator zurück; verwenden Sie async for, um Verzeichniseinträge zu konsumieren.

Verwenden Sie Range Reads für Modellformate, die nicht jedes Byte benötigen. Der verteilte Accelerator für Cloud-Speicher speichert nur die Chunks, die angeforderte Bereiche abdecken.

Leseleistung optimieren

Passen Sie die Anzahl gleichzeitiger Lesevorgänge und die angeforderte Byte-Bereichsgröße gemeinsam an. Die besten Einstellungen hängen von deinem Zugriffsmuster, den Pod-CPU- und Speichergrenzen, der Bandbreite des Client- und Cache-Node-Netzwerks sowie davon ab, ob die Reads den Cache erreichen oder auf Blob Storage zurückfallen.

Verwenden Sie die Werte in diesem Abschnitt als Ausgangspunkt für Workload-Benchmarks, nicht als garantierte Leistungsempfehlungen. Messen Sie deren Einfluss mit Ihrem eigenen Workload, bevor Sie Änderungen bereitstellen.

Parallelität anpassen

Die Anzahl gleichzeitiger Lesevorgänge ist die Anzahl der Lesevorgänge, die deine Anwendung gleichzeitig ausstehend hält. Eine Schleife, die auf jeden read_blob Aufruf wartet, bevor sie den nächsten startet, ist sequentiell, selbst wenn sie die asynchrone API verwendet.

• Gleichzeitig unabhängige Byte-Range- oder Blob-Lesevorgänge ausführen. Begrenzen Sie die Anzahl der aktiven Lesevorgänge mit einem asyncio.Semaphore oder einer festen Anzahl von Worker-Aufgaben. Erstelle nicht für jeden Bereich eines großen Datensatzes gleichzeitig eine Aufgabe.
• Beginne mit einem kleinen Limit, wie vier gleichzeitige Lesevorgänge, und vergleiche es mit acht und sechzehn. Erhöhe die Nebenläufigkeit nur, solange sich der Durchsatz ohne inakzeptable Latenz, Speicherverbrauch, Fehler oder Drosselung verbessert.
• Der native Client akzeptiert ein concurrent.futures.Executor über seinen executor Konstruktorparameter. Um den Python-Threadpool explizit zu kontrollieren, übergebe zum Beispiel ein DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)), ThreadPoolExecutor. Behalte eine Referenz auf den Executor, halte ihn am Leben, bis alle Reads abgeschlossen sind, und schalte ihn nach dem Aufruf von await client.shutdown() ab.
• Ein größerer Threadpool macht sequenzielle Aufrufe nicht parallel. Binde Anwendungsaufgaben sowie Executor-Worker und verwende für jeden aktiven Lesevorgang einen separaten Puffer.

Berücksichtigen Sie die Parallelität über alle Worker-Prozesse und Pods. Eine Erhöhung der Nebenläufigkeit jedes Pods während eines Scale-outs kann dieselben Cache-Knoten überlasten oder doppelte Blob-Downloads bei Cold-Cache-Fehlzugriffen erhöhen.

Konfigurieren Sie die Cache-Chunk-Größe

Setze SI_streamingChunkSize für Anwendungspods in Bytes, um die Cache-Chunk-Größe zu konfigurieren. Zum Beispiel verwendet die folgende Einstellung 3-MiB-Chunks:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Setze diese Variable, bevor du den ersten Client in einem Prozess erstellst. Starte die Application Pods nach der Änderung neu.

Die Standard-Cache-Chunk-Größe beträgt 32 MiB. Das Argument size zu read_blob(path, buffer, size, offset) spezifiziert den von der Anwendung angeforderten Byte-Bereich, nicht die Cache-Chunk-Größe.

Zugriffsmuster	Lesehinweise zur Anwendung
Lesevorgänge des großen sequenziellen Modells	Vergleichen Sie größere Cache-Chunks, angeforderte Bereichsgrößen und begrenzte Parallelität. Ändern Sie jeweils eine Einstellung.
Kleine oder zufällige Lesezugriffe	Fordern Sie nur die Bytes an, die die Anwendung benötigt. Vergleichen Sie kleinere Cache-Chunks, um die außerhalb des angeforderten Bereichs abgerufene Datenmenge zu reduzieren.
Kapazitätsbeschränkte Caches	Messen Sie die Cache-Belegung und das Eviction-Verhalten, anstatt den Cache-Verbrauch nur anhand der Anwendungs-Lesepuffergrößen zu schätzen.

Budget-Anwendung lesen Puffer separat von Cache-Chunks. Beispielsweise erfordern acht gleichzeitige 4-MiB-read_blobAnfragen 32 MiB Lesepufferkapazität der Anwendung, unabhängig von der konfigurierten Cache-Chunk-Größe. Reservieren Sie zusätzlichen Pod-Speicher für native Client-Allokationen, Warteschlangenarbeiten, Python-Objekte und Modellverarbeitung. Puffer erst nach Erreichen des Parallelitätslimits anfordern und jeden Puffer in einem finally-Block freigeben.

Warning

BoundedBufferManager(max_memory_limit=...) begrenzt die Lesepufferkapazität der Anwendung, nicht den gesamten Client-Speicher oder die native Upload-Pufferung. In 0.1.0beträgt das standardmäßige asynchrone Upload-Puffer-Budget des nativen Clients 400 GiB. Dieses Budget ist weder eine Vorallokation noch ein sicheres Pod-Speicherlimit. Große Cold-Cache-Reads können einen kleineren Pod erschöpfen. Für speicherbeschränkte Arbeitslasten testen Sie synchrone Population (SI_cacheEnableSyncUpload=true) oder ein begrenztes asynchrones Upload-Budget (SI_cacheMaxUploadBufferSizePerProcessGB). Legen Sie diese Werte vor der Erstellung des ersten Clients fest, und budgetieren Sie native Puffer getrennt von Anwendungslesepuffern und der Modellverarbeitung.

Messen Sie, bevor Sie die Grenzwerte erhöhen

1. Stellen Sie eine Basislinie mit festem Datensatz, Pod-Anzahl und Ressourcenbegrenzungen fest. Messen Sie Durchsatz, Leselatenz, CPU-, Speicher- und Netzwerknutzung.
2. Benchmark Cold-Cache- und Warm-Cache-Lesungen getrennt. Inspect cache_server_request_counter und Azure Blob-Telemetrie, um Cache-Verkehr von Direct-Blob-Fallback zu unterscheiden.
3. Ändern Sie jeweils eine Einstellung: Nebenläufigkeit, Anwendungslesegröße oder Cache-Chunk-Größe. Halte den Datensatz und andere Einstellungen beim Vergleich der Ergebnisse konsistent.
4. Hören Sie auf, eine Einstellung zu erhöhen, wenn der Durchsatz stagniert oder Latenz, Ressourcendruck, Fehler oder Blob-Throttling zunehmen.

Für cacheseitige Beobachtungen siehe Überwachen des verteilten Beschleunigers für Cloudspeicher.

Fehler behandeln

Alle DACS-Ausnahmen leiten sich von DACSError ab:

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	Dies bedeutet
DACSNotFoundError	Der Blob-Pfad oder das Verzeichnis existiert nicht.
DACSExistsError	Eine Anwendung versucht, in einen bestehenden Blob zu schreiben, ohne dass das Überschreiben aktiviert ist.
DACSAuthenticationError	Anmeldeinformationen fehlen oder sind ungültig.
DACSPermissionError	Der Identität fehlt die erforderliche Blob-Rolle.
DACSTimeoutError	Bei einem Cache- oder Blob-Vorgang ist eine Zeitüberschreitung aufgetreten.
DACSNetworkError	Der Cache-Server oder Blob-Endpunkt ist nicht erreichbar.
DACSInvalidInputError	Ein Pfad oder Argument ist ungültig.
OperationNotSupportedError	Die angeforderte API ist noch nicht implementiert.

Protokollierung steuern

Der Client verfügt über separate Loging-Steuerungen für die Python- und native Schicht. Setze beide Variablen so, dass sie ausführliche Ausgaben aus beiden Schichten unterdrücken:

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

DACS_DATA_LOG_LEVEL steuert das Python-Logging. STORAGE_LOG_LEVEL Steuert die Protokollierung des nativen Clients.

Nächste Schritte

• Verwalten Sie den verteilten Accelerator für Cloud-Speicher.
• Überwachen Sie den verteilten Accelerator für Cloud-Speicher.
• Fehlerbehebung eines verteilten Beschleunigers für Cloud-Speicher.