Utiliza un acelerador distribuido para almacenamiento en la nube con el cliente de Python (vista previa)

El acelerador distribuido para almacenamiento en la nube (DACS) proporciona un cliente nativo de Python para leer datos de Azure Blob a través de la caché distribuida.

Completa Comienza con el acelerador distribuido para almacenamiento en la nube en AKS antes de usar este artículo.

Instala el cliente nativo de Python

Instala el paquete desde PyPI en tu imagen de carga de trabajo:

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

Si estás actualizando una carga de trabajo existente, actualiza el pin del paquete en su definición de compilación de imagen, reconstruye y empuja la imagen, y vuelve a desplegar los pods de carga de trabajo con la imagen actualizada. Instalar el paquete localmente no actualiza los contenedores en ejecución. Para comandos de construcción y push de imagen, véase Añadir el cliente a una imagen de carga de trabajo.

El paquete requiere Python 3.8 o más reciente y un entorno Linux con glibc 2.35 o posterior. Proporciona wheels autocontenidos manylinux_2_35 para x86-64 y arm64, incluyendo la biblioteca nativa y sus dependencias de biblioteca compartida. La azure-identity dependencia se instala automáticamente.

El inicio rápido utiliza Python 3.12 en python:3.12-slim-bookworm con nodos x86-64.

Verifica la instalación dentro del contenedor de carga de trabajo:

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

El comando debe imprimir 0.1.0. El nombre de la distribución es dacs-client, pero el nombre de importación de Python es py_dacs_client.

Configurar el cliente

Establece ambas variables de descubrimiento de caché en cada pod que use acelerador distribuido para almacenamiento en la nube:

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

Importante

Ambas variables son necesarias. Si falta alguno de los dos, el cliente registra Cache not configured. Working in Remote Blob mode only. y accede directamente a Azure Blob Storage. La operación puede tener éxito sin usar la caché.

Utiliza el nombre de host del servicio de descubrimiento para tu recurso y espacio de nombres de caché, por ejemplo, cache-sample-discovery.<namespace>.svc.cluster.local, con puerto 9065.

Configura la autenticación de Blob antes de crear el cliente. El siguiente ejemplo utiliza la identidad gestionada de kubelet de AKS a través del Servicio de Metadatos de Instancia (IMDS) del nodo, no la identidad federada de carga de trabajo de AKS. Establezca AZURE_CLIENT_ID en el ID de cliente de la identidad de 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"

El Pod debe poder llegar a IMDS. Para esta configuración cliente nativo, MSI_SECRET es un marcador de compatibilidad no vacío requerido, no un secreto o credencial real. No pongas una credencial en ese valor. Estos ajustes son específicos de esta integración; no los copies en la configuración de autenticación de Run:AI.

Para inferencia de solo lectura, concede a la identidad de kubelet el rol Storage Blob Data Reader con alcance al contenedor del modelo. Consulta Conceder acceso de lectura a los datos del modelo. Utiliza Storage Blob Data Contributor solo cuando la aplicación deba escribir datos. No se necesita el permiso de escritura de un cargador independiente para las lecturas en tiempo de ejecución.

Utiliza este formato de ruta canónica:

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

La capa nativa lee la configuración cuando se crea el primer cliente en un proceso. Configura el entorno y la configuración del cliente antes de crear ese cliente. Los clientes que se conecten después no cambian la configuración nativa que ya está en vigor.

Entiende los fallos de caché y las lecturas concurrentes

Cuando varios clientes solicitan los mismos datos faltantes al mismo tiempo, cada cliente puede recuperar los datos de forma independiente de Azure Blob Storage. La población coordinada de caché en frío no está disponible en esta vista previa.

La población de la caché es asincrónica. Una lectura puede terminar antes de que sus datos estén disponibles desde la caché, por lo que dos lecturas inmediatas pueden dar fallo de caché incluso cuando los bytes devueltos coinciden.

Antes de eventos a gran escala sincronizados, precaliente los blobs de modelos seleccionados mediante DACS antes de iniciar las réplicas de inferencia. Consulta Precargar datos del modelo antes del despliegue para obtener información sobre un pod de calentamiento dedicado, comandos de lectura y verificación del despliegue.

Una lectura correcta puede provenir de la conmutación por error de Azure Blob Storage. Para verificar que las lecturas usan DACS:

1. Antes de la lectura, registre cache_server_request_counter{request_type="Download",status="Success"}, la suma de todos los pods del servidor de caché para su recurso de caché. Los valores de la etiqueta distinguen entre mayúsculas y minúsculas.
2. Lee repetidamente el mismo rango de blob con un retraso entre lecturas. La guía de inicio rápido utiliza cinco lecturas espaciadas a intervalos de un segundo; esta secuencia no garantiza que la población se haya completado. Compruebe que el recuento de descargas correctas del servidor aumente y que el cliente registre aciertos en la caché. Comprueba los recuentos de bytes y las sumas de comprobación devueltos para asegurar la integridad de los datos; un tiempo más rápido por sí solo no es prueba de caché.
3. Correlacione las métricas del cliente y del servidor con la telemetría de Azure Blob para distinguir el tráfico de caché de las lecturas del origen y del recurso de reserva.

Para comandos que recopilan métricas de cliente y contadores de servidor, véase Verificar la ruta de solicitud DACS.

Usa el cliente nativo

El cliente nativo proporciona métodos asíncronos y sus equivalentes síncronos. Lee datos en búferes nativos preasignados. Establezca MODEL_PATH como la ruta canonical az:// de un fragmento de modelo existente antes de ejecutar el siguiente ejemplo, que lee hasta los primeros 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())

Los métodos de lectura más comunes incluyen:

Método asíncrono	Método síncrono	Description
get_file_size(path)	get_file_size_sync(path)	Obtén el tamaño del blob en bytes.
get_file_stat(path)	get_file_stat_sync(path)	Obtén metadatos del archivo de modelo.
read_blob(path, buffer, size, offset)	read_blob_sync(...)	Lee un rango de bytes en un búfer preasignado.
list_dir(path)	list_dir_sync(path)	Lista de entradas del directorio.
get_file_handle(path, flags)	get_file_handle_sync(...)	Abra un identificador reutilizable.
release_file_handle(handle)	release_file_handle_sync(handle)	Cierre un identificador reutilizable.
shutdown()	shutdown_sync()	Libera conexiones y emite métricas finales.

El método asíncrono list_dir(path) devuelve un iterador; se usa async for para consumir entradas de directorio.

Usa lecturas por rango para formatos de modelo que no necesiten todos los bytes. El acelerador distribuido para almacenamiento en la nube almacena solo los fragmentos que cubren los rangos solicitados.

Optimizar el rendimiento de lectura

Ajuste conjuntamente la concurrencia de lectura de la aplicación y el tamaño del rango de bytes solicitado. La configuración óptima depende de tu patrón de acceso, de los límites de CPU y memoria del pod, del ancho de banda de red del cliente y del nodo de caché, y de si las lecturas se realizan desde la caché o se recurren a Blob Storage.

Utiliza los valores de esta sección como puntos de partida para benchmarks de carga de trabajo, no como recomendaciones de rendimiento garantizadas. Compruebe su efecto con su propia carga de trabajo antes de implementar los cambios.

Ajustar la concurrencia

La concurrencia es el número de lecturas que tu aplicación mantiene en curso. Un bucle que espera a que termine cada llamada a read_blob antes de iniciar la siguiente es secuencial, incluso cuando utiliza la API asíncrona.

• Realizar lecturas independientes de rango de bytes o de blobs simultáneamente. Limita el número de lecturas activas con un asyncio.Semaphore o un número fijo de tareas de trabajo. No crees una tarea para cada rango de un gran conjunto de datos a la vez.
• Empieza con un límite pequeño, como cuatro lecturas concurrentes, y compáralo con ocho y dieciséis. Aumente la concurrencia solo mientras la capacidad de rendimiento mejore sin una latencia, un uso de memoria, errores o limitación de la tasa inaceptables.
• El cliente nativo admite un concurrent.futures.Executor a través de su executor parámetro del constructor. Para controlar explícitamente el pool de hilos de Python, pasa un ThreadPoolExecutor, por ejemplo, DACSCacheClient(executor=ThreadPoolExecutor(max_workers=8)). Conserva una referencia al ejecutor, mantenla activa hasta que terminen todas las lecturas y deténlo tras llamar a await client.shutdown().
• Un pool de hilos más grande no convierte las llamadas secuenciales en concurrentes. Asigne tareas de aplicación y procesos de ejecución, y utilice un búfer independiente para cada lectura activa.

Ten en cuenta la concurrencia en todos los procesos y pods de los trabajadores. Aumentar la concurrencia de cada pod durante un escalamiento puede sobrecargar los mismos nodos de caché o aumentar las descargas duplicadas de Blob en caso de fallos en la caché en frío.

Configurar el tamaño del bloque de caché

Establezca SI_streamingChunkSize en bytes en los pods de aplicación para configurar el tamaño del bloque de caché. Por ejemplo, la siguiente configuración utiliza bloques de 3 MiB:

env:
  - name: SI_streamingChunkSize
    value: "3145728"

Establece esta variable antes de crear el primer cliente en un proceso. Reinicie los pods de aplicación después de realizar el cambio.

El tamaño predeterminado del bloque de caché es de 32 MiB. El argumento size de read_blob(path, buffer, size, offset) especifica el rango de bytes solicitado por la aplicación, no el tamaño del bloque de caché.

Patrón de acceso	Guía de lectura sobre la aplicación
Lecturas de modelo secuencial grande	Compara fragmentos de caché más grandes, tamaños de rango solicitados y concurrencia acotada. Cambie una configuración cada vez.
Lecturas pequeñas o aleatorias	Solicita solo los bytes que la aplicación necesita. Comparar los bloques de caché más pequeños para reducir los datos obtenidos fuera del rango solicitado.
Memorias caché con capacidad limitada	Mida el nivel de ocupación de la caché y el comportamiento de expulsión, en lugar de estimar el uso de la caché basándote únicamente en el tamaño de buffer de lectura de las aplicaciones.

La aplicación Budget lee buffers por separado a partir de bloques de caché. Por ejemplo, ocho solicitudes concurrentes de 4 MiB read_blob requieren 32 MiB de capacidad de búfer de lectura de la aplicación, independientemente del tamaño configurado del bloque de caché. Reserva memoria pod adicional para asignaciones del cliente nativo, trabajo en cola, objetos Python y procesamiento de modelos. Adquiera buffers únicamente después de alcanzar el límite de concurrencia y libere cada buffer en un finally bloque.

Warning

BoundedBufferManager(max_memory_limit=...) limita la capacidad del búfer de lectura de la aplicación, no la memoria total del cliente ni el almacenamiento nativo de carga. En 0.1.0, el presupuesto por defecto del buffer de subida asíncrono del cliente nativo es de 400 GiB. Este presupuesto no es una preasignación ni un límite seguro de memoria Pod. Las lecturas grandes de caché en frío pueden agotar un Pod más pequeño. Para cargas de trabajo con limitaciones de memoria, prueba el llenado síncrono (SI_cacheEnableSyncUpload=true) o un presupuesto limitado para carga asíncrona (SI_cacheMaxUploadBufferSizePerProcessGB). Establece estos valores antes de crear el primer cliente y asigna presupuesto por separado a los buffers nativos, a los buffers de lectura de la aplicación y al procesamiento del modelo.

Medir antes de aumentar los límites

1. Establece una línea base con un conjunto de datos fijo, número de pods y límites de recursos. Mida la capacidad de procesamiento, la latencia de lectura, el uso de CPU, memoria y red.
2. Mida por separado las lecturas con caché fría y caché caliente. Inspeccione cache_server_request_counter y la telemetría de Azure Blob para distinguir el tráfico de caché del fallback direct-Blob.
3. Cambia una configuración a la vez: concurrencia, tamaño de lectura de la aplicación o tamaño del bloque de caché. Mantén el conjunto de datos y otros ajustes consistentes al comparar los resultados.
4. Deja de aumentar un ajuste cuando el rendimiento se estanca o la latencia, la presión de recursos, los errores o la limitación de Blob aumentan.

Para observaciones en el lado de la caché, véase Supervisar el acelerador distribuido para el almacenamiento en la nube.

Gestión de errores

Todas las excepciones de DACS derivan 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)

Exception	Meaning
DACSNotFoundError	La ruta o directorio del blob no existe.
DACSExistsError	Una aplicación intenta escribir en un blob existente sin habilitar la sobrescritura.
DACSAuthenticationError	Faltan credenciales o no son válidas.
DACSPermissionError	La identidad carece del rol de Blob necesario.
DACSTimeoutError	Una caché o una operación de Blob superó el tiempo de espera.
DACSNetworkError	El servidor de caché o el endpoint Blob es inaccesible.
DACSInvalidInputError	La ruta o el argumento tienen un formato no válido.
OperationNotSupportedError	La API solicitada aún no está implementada.

Registro de control

El cliente tiene controles de registro separados para las capas Python y nativa. Establece ambas variables para suprimir la salida verbosa de ambas capas:

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

DACS_DATA_LOG_LEVEL controla el logging de Python. STORAGE_LOG_LEVEL controla el registro nativo del cliente.

Pasos siguientes

• Gestionar el acelerador distribuido para almacenamiento en la nube.
• Supervise el acelerador distribuido para el almacenamiento en la nube.
• Soluciona problemas con el acelerador distribuido para almacenamiento en la nube.