Utilize o acelerador distribuído para armazenamento na cloud com o cliente Python (pré-visualização)

O acelerador distribuído para armazenamento na nuvem (DACS) fornece um cliente Python nativo para ler dados do Azure Blob através da cache distribuída.

Conclua Introdução ao acelerador distribuído para armazenamento na nuvem no AKS antes de utilizar este artigo.

Instale o cliente Python nativo

Instale o pacote do PyPI na imagem da sua carga de trabalho:

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

Se estiveres a atualizar um workload existente, atualiza o pin do pacote na definição da build da imagem, reconstrói e envia a imagem, e reimplementa os pods do workload com a imagem atualizada. Instalar o pacote localmente não atualiza os contentores em execução. Para comandos de build e push de imagem, veja Adicionar o cliente a uma imagem de workload.

O pacote requer Python 3.8 ou mais recente e um ambiente Linux com glibc 2.35 ou mais recente. Fornece wheels autónomas manylinux_2_35 para x86-64 e arm64, incluindo a biblioteca nativa e as suas dependências de bibliotecas partilhadas. A dependência azure-identity é instalada automaticamente.

O quickstart usa Python 3.12 em python:3.12-slim-bookworm com nós x86-64.

Verifique a instalação dentro do contentor da workload:

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

O comando deve imprimir 0.1.0. O nome da distribuição é dacs-client, mas o nome de importação em Python é py_dacs_client.

Configurar o cliente

Defina ambas as variáveis de descoberta de cache em cada pod que utilize acelerador distribuído para armazenamento na cloud:

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

Utilize o nome de anfitrião do serviço de deteção para o seu recurso Cache e espaço de nomes, por exemplo, cache-sample-discovery.<namespace>.svc.cluster.local, com a porta 9065.

Configure a autenticação por Blob antes de criar o cliente. O exemplo seguinte utiliza a identidade gerida do kubelet do AKS através do Serviço de Metadados de Instância (IMDS) do nó, e não a identidade federada da carga de trabalho do AKS. Defina AZURE_CLIENT_ID como o ID do cliente da identidade do 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"

O Pod tem de conseguir chegar ao IMDS. Para esta configuração cliente nativo, MSI_SECRET é um marcador de compatibilidade não vazio obrigatório, não um segredo ou credencial real. Não coloques credenciais nesse campo. Estas definições são específicas para esta integração; não as copie para a configuração de autenticação do Run:ai.

Para inferência apenas de leitura, conceda à identidade do kubelet o papel Storage Blob Data Reader com o âmbito do contentor do modelo. Ver Conceder acesso de leitura aos dados do modelo. Utilize Storage Blob Data Contributor apenas quando a aplicação tiver de escrever dados. Não é necessária a permissão de escrita de um uploader separado para leituras em tempo de execução.

Use este formato de caminho canónico:

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

A camada nativa lê a configuração quando o primeiro cliente é criado num processo. Configure o ambiente e as definições do cliente antes de criar esse cliente. Os clientes posteriores não alteram a configuração nativa já em vigor.

Compreender as falhas na cache e as leituras simultâneas

Quando vários clientes solicitam os mesmos dados em falta ao mesmo tempo, cada cliente pode recuperar os dados de forma independente do Armazenamento de Blobs do Azure. A população coordenada de cold-cache não está disponível nesta pré-visualização.

A população da cache é assíncrona. Uma leitura pode terminar antes de os seus dados estarem disponíveis na cache, pelo que duas leituras imediatas podem resultar em cache miss mesmo quando os bytes devolvidos coincidem.

Antes de grandes eventos em escala sincronizada, aqueça os blobs de modelo selecionados através do DACS antes de iniciar réplicas de inferência. Consulte Pré-carregar dados do modelo antes da implantação para um pod de aquecimento dedicado, comandos de leitura e verificação da implantação.

Uma leitura bem-sucedida pode resultar do mecanismo de contingência do Armazenamento de Blobs do Azure. Para verificar que as leituras usam DACS:

1. Antes da leitura, regista cache_server_request_counter{request_type="Download",status="Success"}, somado por todos os pods do servidor de cache para o teu recurso de Cache. Os valores do rótulo diferenciam maiúsculas de minúsculas.
2. Lê repetidamente o mesmo intervalo de blobs com um atraso entre leituras. O quickstart usa cinco leituras espaçadas por um segundo; esta sequência não garante que a população tenha terminado. Confirme que o número de downloads bem-sucedidos do servidor aumente e que o cliente reporta os acertos na cache. Verifique as contagens de bytes retornados e as somas de verificação para a integridade dos dados; o tempo mais rápido por si só não é prova de cache.
3. Correlacione métricas de cliente e servidor com telemetria Azure Blob para distinguir o tráfego de cache das leituras de origem e do fallback.

Para comandos para recolher métricas do cliente e contadores do servidor, consulte Verificar o caminho do pedido DACS.

Use o cliente nativo

O cliente nativo fornece métodos assíncronos e as respetivas versões síncronas. Lê para buffers nativos pré-alocados. Defina MODEL_PATH para o caminho canónico az:// de um fragmento de modelo existente antes de executar o seguinte exemplo, que lê até aos primeiros 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())

Métodos comuns de leitura incluem:

O método assíncrono list_dir(path) devolve um iterador; use async for para consumir entradas de diretório.

Use leituras por intervalo para formatos de modelo que não precisem de todos os bytes. O acelerador distribuído para armazenamento na cloud armazena apenas os blocos que cobrem os intervalos solicitados.

Otimizar o desempenho de leitura

Ajuste em conjunto a concorrência de leitura da aplicação e o tamanho do intervalo de bytes solicitado. As melhores definições dependem do teu padrão de acesso, limites de CPU e memória do pod, largura de banda da rede do cliente e do nó de cache, e se as leituras atingem a cache ou recorrem ao Armazenamento de Blobs.

Utilize os valores desta secção como pontos de partida para benchmarks de carga de trabalho, não como recomendações de desempenho garantido. Mede o efeito deles com a tua própria carga de trabalho antes de implementares as alterações.

Ajustar concorrência

A concorrência é o número de leituras que a sua aplicação mantém em curso. Um ciclo que aguarda cada chamada read_blob antes de iniciar a seguinte é sequencial, mesmo quando utiliza a API assíncrona.

• Emitir leituras independentes de intervalo de bytes ou blob em simultâneo. Limite o número de leituras ativas com um asyncio.Semaphore ou um número fixo de tarefas de processamento. Não crie uma tarefa para cada intervalo de um grande conjunto de dados de uma só vez.
• Comece com um limite pequeno, como quatro leituras simultâneas, e compare com oito e dezasseis. Aumentar a concorrência apenas enquanto o débito melhora, sem latência, uso de memória, erros ou limitação inaceitáveis.
• O cliente nativo aceita um concurrent.futures.Executor através do seu executor parâmetro construtor. Para controlar explicitamente o pool de threads Python, passe um ThreadPoolExecutor, por exemplo, DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)). Manter uma referência ao executor, mantê-la ativa até todas as leituras terminarem e desligá-la após chamar await client.shutdown().
• Um pool de threads maior não torna chamadas sequenciais concorrentes. Vincule tarefas de aplicação, bem como workers do executor, e use um buffer separado para cada leitura ativa.

Tenha em conta a concorrência em todos os processos de trabalho e pods. Aumentar a concorrência de cada pod durante um scale-out pode sobrecarregar os mesmos nós de cache ou aumentar downloads duplicados de Blob em falhas de cold cache.

Configurar o tamanho do bloco de cache

Defina SI_streamingChunkSize, em bytes, nos pods de aplicação para configurar o tamanho do bloco da cache. Por exemplo, a configuração seguinte usa blocos de 3 MiB:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Defina esta variável antes de criar o primeiro cliente num processo em execução. Reinicie os pods da aplicação depois de fazer alterações.

O tamanho padrão do bloco de cache é de 32 MiB. O argumento size de read_blob(path, buffer, size, offset) especifica o intervalo de bytes solicitado pela aplicação, não o tamanho do bloco da cache.

A aplicação Budget lê buffers separadamente dos blocos de cache. Por exemplo, oito pedidos simultâneos de 4 MiB read_blob requerem 32 MiB de capacidade de buffer de leitura da aplicação, independentemente do tamanho configurado do bloco de cache. Reserve memória pod adicional para alocações nativas do cliente, trabalho em fila, objetos Python e processamento de modelos. Adquira buffers apenas após atingir o limite de concorrência e liberte cada buffer num bloco finally.

Meça antes de aumentar os limites

1. Estabeleça uma linha de base com um conjunto de dados fixo, número de pods e limites de recursos. Medir o débito, a latência de leitura, a CPU, a memória e a utilização da rede.
2. Avalie separadamente as leituras cold-cache e warm-cache. Inspect cache_server_request_counter e a telemetria do Azure Blob para distinguir o tráfego de cache do fallback direto do Blob.
3. Muda uma definição de cada vez: concorrência, tamanho de leitura da aplicação ou tamanho do bloco de cache. Mantenha o conjunto de dados e outras definições consistentes ao comparar resultados.
4. Pare de aumentar uma configuração quando o throughput estagnar ou a latência, a pressão de recursos, os erros ou o Blob throttling aumentarem.

Para observações do lado da cache, veja Monitorizar o acelerador distribuído para armazenamento na cloud.

Lidar com erros

Todas as exceções ao DACS derivam de 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)

Registo de eventos de controlo

O cliente tem controlos de registo separados para as camadas Python e nativa. Defina ambas as variáveis para suprimir a saída verbosa de ambas as camadas:

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

DACS_DATA_LOG_LEVELcontrola o logging em Python. STORAGE_LOG_LEVEL controlos do registo nativo do cliente.

Passos seguintes

• Gerir acelerador distribuído para armazenamento na cloud.
• Monitorizar acelerador distribuído para armazenamento na nuvem.
• Resolver problemas no acelerador distribuído para armazenamento na cloud.