Usar o acelerador distribuído para armazenamento em nuvem com o cliente Python (versão prévia)

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

Conclua Comece com o acelerador distribuído para armazenamento em nuvem no AKS antes de usar este artigo.

Instale o cliente nativo de Python

Instale o pacote do PyPI na imagem de workload:

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

Se você estiver atualizando uma carga de trabalho existente, atualize o pin do pacote na definição da build da imagem, reconstrua e faça push na imagem, e reimplante os pods de carga de trabalho com a imagem atualizada. Instalar o pacote localmente não atualiza os containers em execução. Para comandos de build e push de imagem, veja Adicionar o cliente a uma imagem de carga de trabalho.

O pacote requer Python 3.8 ou mais recente e um ambiente Linux com glibc 2.35 ou mais recente. Ela fornece wheels autônomas manylinux_2_35 para x86-64 e arm64, incluindo a biblioteca nativa e suas dependências de biblioteca compartilhada. A dependência azure-identity se instala automaticamente.

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

Verifique a instalação dentro do contêiner de 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 usa acelerador distribuído para armazenamento em nuvem:

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

Use o nome do host do serviço de descoberta para seu recurso de cache e namespace, por exemplo, cache-sample-discovery.<namespace>.svc.cluster.local, com a porta 9065.

Configure a autenticação do Blob antes de criar o cliente. O exemplo a seguir utiliza a identidade gerenciada do kubelet do AKS através do Serviço de Metadados de Instância (IMDS) do nó, não a identidade federada da carga de trabalho do AKS. Defina AZURE_CLIENT_ID para o ID do cliente da identidade 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 deve conseguir alcançar o IMDS. Para essa configuração de cliente nativo, MSI_SECRET é um marcador de compatibilidade não vazio obrigatório, não um segredo ou credencial real. Não coloque uma credencial nesse valor. Essas configurações são específicas para essa integração; não as copie para a configuração de autenticação do Run:AI.

Para inferência somente leitura, conceda à identidade do kubelet o papel Storage Blob Data Reader com escopo para o contêiner do modelo. Consulte Conceder acesso de leitura aos dados do modelo. Use Colaborador de dados de blob de armazenamento apenas quando o aplicativo precisar gravar dados. Não é necessária a permissão de gravação de um carregador 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 em um processo. Configure o ambiente e as configurações do cliente antes de criar esse cliente. Clientes subsequentes não mudam a configuração nativa já em vigor.

Entenda cache misses e leituras simultâneas

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

A população do cache é assíncrona. Uma leitura pode terminar antes que seus dados estejam disponíveis no cache, então duas leituras imediatas podem resultar em cache miss mesmo quando os bytes retornados coincidem.

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

Uma leitura bem-sucedida pode vir do fallback do Armazenamento de Blobs do Azure. Para verificar se as leituras usam o DACS:

1. Antes de realizar a leitura, registre cache_server_request_counter{request_type="Download",status="Success"}, somado em todos os pods de servidor de cache para seu recurso de Cache. Os valores de rótulo diferenciam maiúsculas de minúsculas.
2. Leia o mesmo intervalo de blob repetidamente com um atraso entre as leituras. O guia de início rápido usa cinco leituras espaçadas por um segundo; essa sequência não garante que o preenchimento tenha sido concluído. Confirme que a contagem de downloads do servidor bem-sucedido aumenta e que o cliente reporta hits de cache. Verifique a contagem de bytes retornados e os checksums para integridade dos dados; tempo mais rápido sozinho não é prova de cache.
3. Correlacione métricas de cliente e servidor com a telemetria do Azure Blob para distinguir o tráfego de cache das leituras de origem e do fallback.

Para comandos de coleta de métricas do cliente e contadores de servidor, veja Verificar o caminho da requisição DACS.

Use o cliente nativo

O cliente nativo fornece métodos assíncronos e contrapartes síncronas. Ele 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é os 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) retorna um iterador; use async for para consumir entradas de diretório.

Use leituras por intervalo para formatos de modelo que não precisam de todos os bytes. Acelerador distribuído para armazenamento em nuvem armazena apenas os blocos que cobrem os intervalos solicitados.

Otimizar o desempenho de leitura

Ajuste a concorrência de leitura do aplicativo e o tamanho do intervalo de bytes solicitado em conjunto. As melhores configurações dependem do seu 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 o cache ou recorrem ao Armazenamento de Blobs.

Use os valores desta seção como pontos de partida para benchmarks de carga de trabalho, não como recomendações de desempenho garantido. Meça o efeito deles com sua própria carga de trabalho antes de implantar mudanças.

Ajustar a concorrência

Concorrência é o número de leituras que sua aplicação mantém em andamento. Um loop que aguarda cada chamada a read_blob antes de iniciar a próxima é sequencial, mesmo quando usa a API assíncrona.

• Emita leituras independentes de intervalo de bytes ou blob simultaneamente. 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 vez.
• Comece com um limite pequeno, como quatro leituras simultâneas, e compare com oito e dezesseis. Aumente a concorrência apenas enquanto a taxa de transferência melhora, sem latência, uso de memória, erros ou limitação de taxa inaceitáveis.
• O cliente nativo aceita concurrent.futures.Executor por meio de seu parâmetro construtor executor. 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é que todas as leituras terminem 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 assim como workers do executor, e use um buffer separado para cada leitura ativa.

Leve em conta a concorrência entre 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 de cache. Por exemplo, a configuração a seguir usa blocos de 3 MiB:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Defina essa variável antes de criar o primeiro cliente no processo. Reinicie os pods de aplicativo depois de alterá-lo.

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

O aplicativo de orçamento lê buffers separadamente das partes de cache. Por exemplo, oito solicitações read_blob simultâneas de 4 MiB requerem 32 MiB de capacidade de buffer de leitura de aplicação, independentemente do tamanho configurado do bloco de cache. Reserve memória adicional do pod para alocações do cliente nativo, trabalho em fila, objetos Python e processamento de modelos. Adquira buffers somente após entrar na seção limitada pela simultaneidade e libere cada buffer em um bloco finally.

Meça antes de aumentar os limites

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

Para observações do lado do cache, confira Monitorar o acelerador distribuído de armazenamento em nuvem.

Lidar com erros

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

Controlar o registro em log

O cliente possui controles de registro 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_LEVEL controla o registro em log do Python. STORAGE_LOG_LEVEL Controla o registro nativo do cliente.

Próximas Etapas 

• Gerencie o acelerador distribuído para armazenamento em nuvem.
• Monitorar o acelerador distribuído para armazenamento em nuvem.
• Solucionar problemas com aceleradores distribuídos para Cloud Storage.