Usa un acceleratore distribuito per lo storage cloud con il client Python (anteprima)

Distributed accelerator for cloud storage (DACS) fornisce un client Python nativo per la lettura dei dati Azure Blob attraverso la cache distribuita.

Completa Inizia con l'acceleratore distribuito per lo storage cloud su AKS prima di utilizzare questo articolo.

Installa il client Python nativo

Installa il pacchetto da PyPI nell'immagine del carico di lavoro:

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

Se stai aggiornando un carico di lavoro esistente, aggiorna il pin del pacchetto nella definizione della build dell'immagine, ricostruisci e spingi l'immagine, e ridistribuisci i pod di carico di lavoro con l'immagine aggiornata. Installare il pacchetto localmente non aggiorna i container in esecuzione. Per i comandi di build e push dell'immagine, vedi Aggiungi il client a un'immagine del carico di lavoro.

Il pacchetto richiede Python 3.8 o più recente e un ambiente Linux con glibc 2.35 o versioni successive. Fornisce wheel autonome manylinux_2_35 per x86-64 e arm64, inclusa la libreria nativa e le sue dipendenze da librerie condivise. La azure-identity dipendenza si installa automaticamente.

Il quickstart utilizza Python 3.12 su python:3.12-slim-bookworm con nodi x86-64.

Verifica l'installazione all'interno del container del workload:

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

Il comando deve stampare 0.1.0. Il nome della distribuzione è dacs-client, ma il nome di importazione Python è py_dacs_client.

Configurare il client

Imposta entrambe le variabili di scoperta della cache su ogni pod che utilizza un acceleratore distribuito per lo storage cloud:

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

Importante

Entrambe le variabili sono necessarie. Se manca uno dei due, il client registra Cache not configured. Working in Remote Blob mode only. nel log e accede direttamente ad Archiviazione BLOB di Azure. L'operazione può avere successo senza utilizzare la cache.

Usa il nome host del servizio di scoperta per la tua risorsa e namespace della cache, ad esempio, cache-sample-discovery.<namespace>.svc.cluster.local, con porta 9065.

Configura l'autenticazione Blob prima di creare il client. Il seguente esempio utilizza l'identità gestita di AKS kubelet tramite l'Instance Metadata Service (IMDS) del nodo, non l'identità federata del carico di lavoro AKS. Imposta AZURE_CLIENT_ID sull'ID client dell'identità kubelet:

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"

Il Pod deve essere in grado di raggiungere IMDS. Per questa configurazione client-nativo, MSI_SECRET è un marcatore di compatibilità non vuoto richiesto, non un vero segreto o credenziali. Non mettere una credenziale in quel valore. Queste impostazioni sono specifiche dell'integrazione; non copiarle nella configurazione di autenticazione di Run:ai.

Per l'inferenza di sola lettura, concedere all'identità kubelet il ruolo Storage Blob Data Reader con ambito al contenitore del modello. Vedi Concedere l'accesso in lettura ai dati del modello. Usa Storage Blob Data Contributor solo quando l'applicazione deve scrivere dati. Non è necessario il permesso di scrittura di un uploader separato per le letture in fase di esecuzione.

Usa questo formato canonico di percorso:

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

Il livello nativo legge la configurazione quando viene creato il primo client in un processo. Configura l'ambiente e le impostazioni del client prima di creare quel client. I client successivi non cambiano la configurazione nativa già in vigo.

Comprendere i cache miss e le letture concorrenti

Quando più client richiedono contemporaneamente gli stessi dati mancanti, ogni client può recuperare autonomamente i dati da Archiviazione BLOB di Azure. La popolazione coordinata del cold cache non è disponibile in questa anteprima.

La popolazione della cache è asincrona. Una lettura può terminare prima che i suoi dati siano disponibili dalla cache, quindi due letture immediate possono entrambe risultare cache miss anche se i byte restituiti corrispondono.

Prima di eventi di scalabilità sincronizzata di grandi dimensioni, precaricare i BLOB di modello selezionati tramite DACS prima di iniziare le repliche di inferenza. Consultare Prelettura dei dati del modello prima della distribuzione per un pod di riscaldamento dedicato, i comandi di lettura e la verifica dell'implementazione.

Una lettura di successo può derivare dal fallback di Archiviazione BLOB di Azure. Per verificare che le letture utilizzino DACS:

1. Prima di leggere, registrare cache_server_request_counter{request_type="Download",status="Success"}, sommato su tutti i pod del server cache per la risorsa cache. I valori delle etichette sono sensibili a maiuscole e minuscole.
2. Leggi ripetutamente lo stesso intervallo di blob con un ritardo tra una lettura e l'altra. Il quickstart utilizza cinque letture distanziate di un secondo; questa sequenza non garantisce che il popolamento sia completato. Conferma che il conteggio dei download riusciti dal server aumenti e che il client segnali gli hit della cache. Controlla il conteggio dei byte restituiti e i checksum per l'integrità dei dati; il tempo più veloce da solo non dimostra l'uso della cache.
3. Correlare le metriche client e server con la telemetria di BLOB di Azure per distinguere il traffico della cache dalle letture di origine e dal fallback.

Per i comandi per raccogliere le metriche del client e i contatori del server, consulta Verifica il percorso delle richieste DACS.

Usa il client nativo

Il client nativo fornisce metodi asincroni e controparti sincrone. Legge in buffer nativi preallocati. Impostare MODEL_PATH sul percorso canonico az:// di una partizione del modello esistente prima di eseguire il seguente esempio, che legge fino ai primi 4 MiB:

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

I metodi di lettura più comuni includono:

Metodo asincrono	Metodo sincrono	Description
get_file_size(path)	get_file_size_sync(path)	Ottieni la dimensione del blob in byte.
get_file_stat(path)	get_file_stat_sync(path)	Ottenere i metadati del file del modello.
read_blob(path, buffer, size, offset)	read_blob_sync(...)	Leggere un intervallo di byte in un buffer preallocato.
list_dir(path)	list_dir_sync(path)	Elencare le voci della directory.
get_file_handle(path, flags)	get_file_handle_sync(...)	Apri un handle riutilizzabile.
release_file_handle(handle)	release_file_handle_sync(handle)	Chiudi un handle riutilizzabile.
shutdown()	shutdown_sync()	Rilascia le connessioni ed emetti le metriche finali.

Il metodo asincrono list_dir(path) restituisce un iteratore; usa async for per consumare le voci della directory.

Usa le letture a intervallo per i formati di modello che non richiedono ogni byte. L'acceleratore distribuito per l'archiviazione cloud memorizza solo i blocchi che coprono gli intervalli richiesti.

Ottimizza le prestazioni di lettura

Ottimizzare congiuntamente la concorrenza di lettura dell'applicazione e la dimensione dell'intervallo di byte richiesto. Le impostazioni migliori dipendono dal modello di accesso, dai limiti di CPU e memoria del pod, dalla larghezza di banda di rete del client e del nodo di cache e dal fatto che le letture raggiungano la cache o eseguano il fallback in Archiviazione BLOB.

Usa i valori in questa sezione come punti di partenza per i benchmark del carico di lavoro, non come raccomandazioni di prestazioni garantite. Misura il loro effetto con il tuo carico di lavoro prima di distribuire i cambiamenti.

Regolare la concorrenza

La concorrenza è il numero di letture che l'applicazione mantiene in esecuzione. Un ciclo che attende ogni chiamata read_blob prima di iniziare la successiva è sequenziale, anche quando utilizza l'API asincrona.

• Esegui letture indipendenti di intervallo di byte o di BLOB contemporaneamente. Limita il numero di letture attive con un asyncio.Semaphore o un numero fisso di task di lavoro. Non creare un compito per ogni intervallo di un grande dataset contemporaneamente.
• Inizia con un piccolo limite, come quattro letture concorrenti, e confrontalo con otto e sedici. Aumentare la concorrenza solo mentre la capacità effettiva migliora senza latenza, uso di memoria, errori o limitazione inaccettabili.
• Il client nativo accetta un concurrent.futures.Executor tramite il parametro del costruttore executor. Per controllare esplicitamente il pool di thread Python, passa un ThreadPoolExecutor, ad esempio, DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)). Conservare un riferimento all'executor, lasciarlo attivo fino a quando tutte le letture non terminano, e arrestarlo dopo aver chiamato await client.shutdown().
• Un pool di thread più grande non rende concorrenti le chiamate sequenziali. Assegnare le attività dell'applicazione così come i worker executor, e utilizzare un buffer separato per ogni lettura attiva.

Tieni conto della concorrenza tra tutti i processi worker e i pod. Aumentare la concorrenza di ogni pod durante una scale-out può sovraccaricare gli stessi nodi di cache o aumentare i download duplicati di Blob in caso di cache miss con cache fredda.

Configura la dimensione del chunk della cache

Imposta SI_streamingChunkSize in byte sui pod applicativi per configurare la dimensione del chunk della cache. Ad esempio, la seguente impostazione utilizza chunk da 3 MiB:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Imposta questa variabile prima di creare il primo client in un processo. Riavviare i pod dell'applicazione dopo averli modificati.

La dimensione predefinita dei blocchi della cache è 32 MiB. L'argomento size impostato su read_blob(path, buffer, size, offset) specifica l'intervallo di byte richiesto dall'applicazione, non la dimensione del blocco della cache.

Modello di accesso	Linee guida per la lettura di applicazioni
Letture del modello sequenziali di grandi dimensioni	Confronta i blocchi di cache più grandi, le dimensioni degli intervalli richiesti e la concorrenza limitata. Modificare un'impostazione alla volta.
Letture piccole o casuali	Richiedi solo i byte di cui l'applicazione ha bisogno. Confrontare i blocchi di cache di dimensioni inferiori per ridurre i dati recuperati fuori dall'intervallo richiesto.
Cache a vincoli di capacità	Misura il comportamento di occupazione e di espulsione della cache piuttosto che stimare l'uso della cache solo in base alle dimensioni del buffer di lettura dell'applicazione.

L'applicazione Budget legge i buffer separatamente dai chunk della cache. Ad esempio, otto richieste contemporanee da 4 MiB read_blob richiedono una capacità del buffer di lettura dell'applicazione di 32 MiB, indipendentemente dalla dimensione configurata del blocco della cache. Riserva memoria pod aggiuntiva per allocazioni native client, lavoro in coda, oggetti Python e elaborazione modello. Acquisire i buffer solo dopo essere entrati nel limite di concorrenza e liberare ogni buffer in un blocco finally.

Avvertimento

BoundedBufferManager(max_memory_limit=...) limita la capacità del buffer di lettura dell'applicazione, non la memoria totale del client o il buffering nativo di upload. In 0.1.0, il budget predefinito del buffer asincrono di upload del client nativo è di 400 GiB. Questo budget non è una preallocazione né un limite sicuro di memoria Pod. Grandi letture di cold cache possono esaurire un Pod più piccolo. Per carichi di lavoro con vincoli di memoria, testare il popolamento sincrono (SI_cacheEnableSyncUpload=true) o un budget di upload asincrono limitato (SI_cacheMaxUploadBufferSizePerProcessGB). Impostare questi valori prima di creare il primo client e prevedere separatamente i buffer nativi dai buffer di lettura dell'applicazione e dall'elaborazione del modello.

Misura prima di aumentare i limiti

1. Definire una baseline con un dataset fisso, il conteggio dei pod e i limiti di risorse. Misura il throughput, la latenza di lettura, la CPU, la memoria e l'uso della rete.
2. Esegui il benchmark delle letture cold-cache e warm-cache separatamente. Esamina cache_server_request_counter e la telemetria di Azure Blob per distinguere il traffico della cache dal fallback diretto al Blob.
3. Cambia un'impostazione alla volta: concorrenza, dimensione di lettura dell'applicazione o dimensione del blocco della cache. Mantieni coerenti il dataset e le altre impostazioni quando confronti i risultati.
4. Smetti di aumentare un'impostazione quando il throughput si stabilizza o la latenza, la pressione delle risorse, gli errori o il Blob throttling aumentano.

Per osservazioni lato cache, vedi Monitorare l'acceleratore distribuito per lo storage cloud.

Gestire gli errori

Tutte le eccezioni DACS derivano da DACSError:

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	Il percorso o la directory del Blob non esistono.
DACSExistsError	Un'applicazione cerca di scrivere su un blob esistente senza la sovrascrittura abilitata.
DACSAuthenticationError	Le credenziali sono mancanti o non valide.
DACSPermissionError	L'identità manca del ruolo richiesto di Blob.
DACSTimeoutError	Una cache o un'operazione BLOB è scaduta per timeout.
DACSNetworkError	Il server cache o l'endpoint Blob sono irraggiungibili.
DACSInvalidInputError	Un percorso o un argomento è malformato.
OperationNotSupportedError	L'API richiesta non è ancora implementata.

Log di controllo

Il client dispone di controlli di logging separati per i livelli Python e nativo. Imposta entrambe le variabili per sopprimere l'output verboso da entrambi i livelli:

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

DACS_DATA_LOG_LEVEL controlla il logging di Python. STORAGE_LOG_LEVEL controlla la registrazione dei log del client nativo.

Passaggi successivi

• Gestisci un acceleratore distribuito per lo storage cloud.
• Monitorare un acceleratore distribuito per l'archiviazione cloud.
• Risoluzione dei problemi con acceleratore distribuito per lo storage cloud.