Użyj rozproszonego akceleratora do przechowywania w chmurze z klientem Python (podgląd)

Rozproszony akcelerator do przechowywania w chmurze (DACS) udostępnia natywnego klienta Python do odczytu danych Azure Blob przez rozproszoną pamięć podręczną.

Ukończ Rozpocznij korzystanie z rozproszonego akceleratora do przechowywania w chmurze na AKS przed użyciem tego artykułu.

Zainstaluj natywny klient Pythona

Zainstaluj pakiet z PyPI w obrazie obciążenia roboczego:

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

Jeśli aktualizujesz istniejące obciążenie robocze, zaktualizuj pin pakietu w definicji budowania obrazu, odbuduj i wypchnij obraz, a następnie ponownie wdroż pody obciążenia roboczego z tym zaktualizowanym obrazem. Instalacja pakietu lokalnie nie aktualizuje działających kontenerów. Aby uzyskać polecenia budowania i wypychania obrazu, zobacz Dodaj klienta do obrazu obciążenia roboczego.

Pakiet wymaga Python 3.8 lub nowszego oraz środowiska Linux z glibc 2.35 lub nowszym. Zapewnia samowystarczalne manylinux_2_35 wheels dla x86-64 i arm64, w tym natywną bibliotekę i jej zależności bibliotek współdzielonych. Zależność azure-identity instaluje się automatycznie.

Quickstart korzysta z Python 3.12 na python:3.12-slim-bookworm z węzłami x86-64.

Sprawdź instalację wewnątrz kontenera obciążenia roboczego:

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

Polecenie musi wypisać 0.1.0. Nazwa dystrybucji to dacs-client, ale import w Python to py_dacs_client.

Konfigurowanie klienta

Ustaw obie zmienne cache-discovery na każdym podzie korzystającym z rozproszonego akceleratora dla pamięci masowej w chmurze:

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

Użyj nazwy hosta usługi discovery dla swojego zasobu cache i przestrzeni nazw, na przykład cache-sample-discovery.<namespace>.svc.cluster.local, z portem 9065.

Skonfiguruj uwierzytelnianie obiektu Blob przed utworzeniem klienta. Poniższy przykład wykorzystuje zarządzaną tożsamość AKS kubelet poprzez usługę Instance Metadata Service (IMDS) węzła, a nie federowaną tożsamość obciążenia AKS. Ustaw AZURE_CLIENT_ID na identyfikator klienta tożsamości kubeleta:

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 musi być w stanie uzyskać dostęp do IMDS. W tej konfiguracji klienta natywnego MSI_SECRET jest wymaganym niepustym znacznikiem zgodności, a nie prawdziwym sekretem ani poświadczeniem. Nie umieszczaj danych uwierzytelniających w tej wartości. Te ustawienia są specyficzne dla tej integracji; nie kopiuj ich do konfiguracji uwierzytelniania Run:ai.

Dla wnioskowania tylko do odczytu należy przypisać tożsamości kubeleta rolę Storage Blob Data Reader w zakresie kontenera modelu. Zobacz Przyznawanie dostępu do odczytu danych modelu. Używaj Storage Blob Data Contributor tylko wtedy, gdy aplikacja musi zapisywać dane. Osobne uprawnienia uploadującego do zapisu nie są potrzebne do odczytu w czasie działania.

Użyj tego kanonicznego formatu ścieżki:

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

Natywna warstwa odczytuje konfigurację, gdy tworzony jest pierwszy klient w procesie. Skonfiguruj środowisko i ustawienia klienta przed utworzeniem tego klienta. Późniejsi klienci nie zmieniają już obowiązującej konfiguracji natywnej.

Zrozumienie chybień pamięci podręcznej i jednoczesnych odczytów

Gdy wielu klientów żąda tych samych brakujących danych jednocześnie, każdy klient może niezależnie pobrać dane z Azure Blob Storage. Skoordynowana populacja zimnych cache'ów nie jest dostępna w tej wersji zapoznawczej.

Populacja pamięci podręcznej jest asynchroniczna. Odczyt może zakończyć się zanim dane będą dostępne z pamięci podręcznej, więc dwa natychmiastowe odczyty mogą chybić nawet wtedy, gdy zwrócone bajty się zgadzają.

Przed dużymi zsynchronizowanymi zdarzeniami skalowania wstępnie załaduj wybrane bloby modelu przez DACS przed rozpoczęciem replik inferencji. Postępuj zgodnie z instrukcjami w Pobierz z wyprzedzeniem dane modelu przed wdrożeniem, aby uzyskać dedykowany pod rozgrzewkowy, zapoznać się z poleceniami odczytu i zweryfikować wdrożenie.

Udany odczyt może pochodzić dzięki mechanizmowi awaryjnemu Azure Blob Storage. Aby sprawdzić, czy odczyty używają DACS:

1. Przed przeczytaniem zapisz cache_server_request_counter{request_type="Download",status="Success"}, zsumowane we wszystkich podach serwera pamięci podręcznej dla Twojego zasobu Cache. Wartości etykiet rozróżniają wielkość liter.
2. Odczytuj ten sam zakres blobów wielokrotnie, z opóźnieniem między odczytami. Quickstart używa pięciu odczytów odstępowanych co sekundę; ta sekwencja nie gwarantuje, że uzupełnianie danych zostało zakończone. Potwierdź, że liczba udanych pobrań serwera rośnie i że klient zgłasza trafienia do pamięci podręcznej. Sprawdź liczbę zwróconych bajtów i sumy kontrolne pod kątem integralności danych; sam krótszy czas wykonania nie jest dowodem na buforowanie.
3. Koreluj metryki klienta i serwera za pomocą telemetrii Azure Blob, aby odróżnić ruch pamięci podręcznej od odczytów z origin oraz trybu awaryjnego.

Polecenia do zbierania metryk klienta i liczników serwera można znaleźć w artykule Verify the DACS request path.

Użyj natywnego klienta

Natywny klient zapewnia metody asynchroniczne oraz ich synchroniczne odpowiedniki. Odczytuje dane do wstępnie przydzielonych natywnych buforów. Ustaw MODEL_PATH na kanoniczną ścieżkę do az:// istniejącego fragmentu modelu przed uruchomieniem następującego przykładu, który odczytuje do pierwszych 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())

Do popularnych metod odczytu należą:

Metoda asynchroniczna list_dir(path) zwraca iterator; użyj async for, aby iterować po wpisach katalogu.

Używaj odczytów zakresu dla formatów modeli, które nie wymagają każdego bajtu. Rozproszony akcelerator pamięci masowej w chmurze przechowuje tylko te fragmenty obejmujące żądane zakresy.

Optymalizuj wydajność odczytu

Dostosuj współbieżność odczytu aplikacji i rozmiar żądanego zakresu bajtów razem. Najlepsze ustawienia zależą od wzorca dostępu, limitów CPU i pamięci pod, przepustowości sieci klienta i węzła podręcznego oraz tego, czy odczyty trafiają do pamięci podręcznej, czy są kierowane do Blob Storage w razie braku trafienia w pamięci podręcznej.

Używaj wartości z tej sekcji jako punktów wyjścia do benchmarków obciążenia, a nie gwarantowanych zaleceń dotyczących wydajności. Mierz ich wpływ na podstawie własnego obciążenia przed wdrożeniem zmian.

Dostosuj współbieżność

Współbieżność to liczba odczytów, które aplikacja utrzymuje w toku. Pętla oczekująca na każde read_blob wywołanie przed rozpoczęciem kolejnego jest sekwencyjna, nawet gdy korzysta z asynchronicznego API.

• Wykonuj niezależne odczyty zakresów bajtów lub obiektów blob równocześnie. Ogranicz liczbę aktywnych odczytów za pomocą asyncio.Semaphore lub stałej liczby zadań roboczych. Nie twórz naraz zadania dla wszystkich zakresów dużego zbioru danych.
• Zacznij od małego limitu, na przykład czterech równoczesnych odczytów, i porównaj go z ośmioma i szesnastoma. Zwiększać współbieżność tylko wtedy, gdy przepustowość się poprawia bez nieakceptowalnych opóźnień, zużycia pamięci, błędów czy dławienia.
• Natywny klient akceptuje concurrent.futures.Executor za pomocą parametru konstruktora executor. Aby wyraźnie kontrolować pulę wątków Python, przekaż ThreadPoolExecutor, na przykład DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)). Zachowaj referencję do executora, utrzymuj go aktywnym do zakończenia wszystkich odczytów, a następnie zamknij go po wywołaniu await client.shutdown().
• Większa pula wątków nie sprawia, że sekwencyjne wywołania stają się współbieżne. Przypisz zadania aplikacji oraz wątków wykonawczych i używaj osobnego bufora dla każdego aktywnego odczytu.

Uwzględnij współbieżność we wszystkich procesach roboczych i podach. Zwiększenie współbieżności każdego podu podczas skalowania w poziomie może przeciążyć te same węzły pamięci podręcznej lub zwiększyć liczbę pobrań obiektów Blob przy chybieniach zimnej pamięci podręcznej.

Konfiguruj rozmiar fragmentu pamięci podręcznej

Ustaw SI_streamingChunkSize w bajtach w podach aplikacji, aby skonfigurować rozmiar fragmentu pamięci podręcznej. Na przykład poniższe ustawienie wykorzystuje bloki 3 MiB:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Ustaw tę zmienną przed utworzeniem pierwszego klienta w procesie. Po zmianie ponownie uruchom pody aplikacji.

Domyślny rozmiar fragmentu pamięci podręcznej to 32 MiB. Argument size do read_blob(path, buffer, size, offset) określa żądany przez aplikację zakres bajtów, a nie rozmiar fragmentu pamięci podręcznej.

Aplikacja Budget odczytuje bufory oddzielnie od fragmentów pamięci podręcznej. Na przykład osiem równoczesnych żądań 4-MiB read_blob wymaga 32 MiB pojemności bufora odczytu aplikacji, niezależnie od skonfigurowanego rozmiaru fragmentu pamięci podręcznej. Zarezerwuj dodatkową pamięć poda na alokacje pamięci klienta natywnego, kolejkowe zadania, obiekty Python oraz przetwarzanie modeli. Pobieraj bufory dopiero po osiągnięciu limitu współbieżności i zwalniaj każdy bufor w bloku finally.

Zmierz przed podniesieniem limitów

1. Ustal punkt odniesienia z ustalonym zbiorem danych, liczbą podów i limitami zasobów. Mierz przepustowość, opóźnienia odczytu, CPU, pamięć i wykorzystanie sieci.
2. Testuj odczyty cold-cache i warm-cache osobno. Sprawdź cache_server_request_counter i telemetrię Azure Blob, aby rozróżnić ruch pamięci podręcznej od awaryjnego przejścia do bezpośredniego dostępu do obiektu Blob.
3. Zmieniaj jedno ustawienie na raz: współbieżność, rozmiar odczytu aplikacji lub rozmiar fragmentu pamięci podręcznej. Utrzymuj spójność zestawu danych i innych ustawień podczas porównywania wyników.
4. Przestań zwiększać ustawienie, gdy przepustowość przestaje rosnąć lub zwiększają się opóźnienia, obciążenie zasobów, błędy lub dławienie usługi Blob.

Informacje dotyczące pamięci podręcznej można znaleźć w artykule Monitorowanie akceleratora rozproszonego dla magazynu w chmurze.

Zarządzanie błędami

Wszystkie wyjątki DACS wynikają z 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)

Sterowanie rejestrowaniem

Klient ma osobne ustawienia rejestrowania dla warstw Python i natywnych. Ustaw obie zmienne, aby tłumić szczegółowe dane wyjściowe z obu warstw:

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

Element DACS_DATA_LOG_LEVEL steruje logowaniem w Pythonie. STORAGE_LOG_LEVEL steruje rejestrowaniem klientów natywnych.

Następne kroki

• Zarządzanie rozproszonym akceleratorem dla pamięci masowej w chmurze.
• Monitor rozproszonego akceleratora dla pamięci masowej w chmurze.
• Rozwiązywanie problemów z rozproszonym akceleratorem do przechowywania w chmurze.