Configuración del cargador automático para cargas de trabajo de producción

Artículo
11/07/2024

Databricks recomienda seguir los procedimientos recomendados de streaming para ejecutar el cargador automático en producción.

Databricks recomienda usar el cargador automático en Delta Live Tables para la ingesta incremental de datos. Delta Live Tables amplía la funcionalidad en Structured Streaming de Apache Spark y le permite escribir solo unas pocas líneas de Python o SQL declarativo para implementar una canalización de datos con calidad de producción con:

Escalado automático de la infraestructura de proceso para ahorrar costes
Controles de calidad de datos con expectativas
Gestión automática de la evolución de esquemas
Supervisión mediante métricas en el registro de eventos

Supervisión del cargador automático

Consulta de archivos detectados por el cargador automático

Nota:

La opción cloud_files_state está disponible en Databricks Runtime 11.3 LTS y versiones posteriores.

El cargadpr automático proporciona una API de SQL para inspeccionar el estado de una secuencia. Con la función cloud_files_state, puede encontrar metadatos sobre los archivos detectados por una secuencia del cargador automático. Basta con consultar desde cloud_files_state, proporcionando la ubicación del punto de control asociada a un flujo del cargador automático.

SELECT * FROM cloud_files_state('path/to/checkpoint');

Escuchar actualizaciones de secuencias

Para supervisar mejor las secuencias de Auto Loader, Databricks recomienda usar la interfaz del cliente de Streaming Query Listener de Apache Spark.

El cargador automático notifica las métricas al cliente de escucha de consultas de streaming en cada lote. Puede ver cuántos archivos existen en el trabajo pendiente y el tamaño este último en las métricas numFilesOutstanding y numBytesOutstanding en la pestaña Datos sin procesar del panel de progreso de la consulta de secuencia:

{
  "sources" : [
    {
      "description" : "CloudFilesSource[/path/to/source]",
      "metrics" : {
        "numFilesOutstanding" : "238",
        "numBytesOutstanding" : "163939124006"
      }
    }
  ]
}

En Databricks Runtime 10.4 LTS y versiones superiores, al usar el modo de notificación de archivos, las métricas también incluirán el número aproximado de eventos de archivo que están en la cola en la nube como approximateQueueSize para AWS y Azure.

Consideraciones sobre el costo

Al ejecutar el cargador automático, el origen principal de los costos sería el costo de los recursos de proceso y la detección de archivos.

Para reducir los costes de proceso, Databricks recomienda usar trabajos de Databricks para programar el cargador automático como trabajos por lotes mediante Trigger.AvailableNow en lugar de ejecutarlo continuamente, siempre que no tenga requisitos de latencia baja. Consulte Configurar intervalos del desencadenador de flujo estructurado.

Los costos de detección de archivos pueden presentarse en forma de operaciones LIST en sus cuentas de almacenamiento en modo de lista de directorios, en forma de solicitudes API en el servicio de suscripción y en forma de servicio de cola en el modo de notificación de archivos. Para reducir los costos de detección de archivos, Databricks recomienda:

Proporcionar un desencadenador ProcessingTime al ejecutar el cargador automático continuamente en el modo de lista de directorios
Arquitectura de las cargas de archivos a su cuenta de almacenamiento en orden léxico para aprovechar el listado incremental (obsoleto) cuando sea posible
Aprovechar las notificaciones de archivos cuando no es posible hacer una lista incremental
Usar etiquetas de recursos para etiquetar los recursos creados por el cargador automático y hacer un seguimiento de los costos

Uso de Trigger.AvailableNow y limitación de velocidad

Nota:

Disponible en Databricks Runtime 10.4 LTS y versiones posteriores.

El cargador automático se puede programar para ejecutarse en trabajos de Databricks como un trabajo por lotes mediante Trigger.AvailableNow. El desencadenador AvailableNow indicará al cargador automático que procese todos los archivos que llegaron antes de la hora de inicio de la consulta. Los nuevos archivos que se cargan después de que se haya iniciado la secuencia se omiten hasta el siguiente desencadenador.

Con Trigger.AvailableNow, la detección de archivos se lleva a cabo de forma asincrónica con el procesamiento de datos y los datos se pueden procesar en varios micro lotes con limitación de velocidad. El cargador automático procesa de manera predeterminada un máximo de 1000 archivos por micro lote. Puede configurar cloudFiles.maxFilesPerTrigger y cloudFiles.maxBytesPerTrigger para configurar cuántos archivos o cuántos bytes se deben procesar en un micro lote. El límite de archivos es un límite máximo, pero el límite de bytes es un límite flexible, lo que significa que se pueden procesar más bytes que el maxBytesPerTrigger proporcionado. Cuando ambas opciones se proporcionan juntas, el cargador automático procesa tantos archivos como sean necesarios para alcanzar uno de los límites.

Ubicación del punto de control

Databricks recomienda establecer la ubicación del punto de control en una ubicación sin una directiva de ciclo de vida de objetos en la nube. Si los archivos de la ubicación del punto de control se limpian según la directiva, el estado de la secuencia está dañado. Si esto sucede, debe reiniciar la secuencia desde cero.

Retención de eventos

El cargador automático hace un seguimiento de los archivos detectados en la ubicación del punto de control mediante RocksDB para proporcionar garantías de ingesta exactamente una vez. Databricks recomienda encarecidamente usar la opción cloudFiles.maxFileAge para todos los flujos de ingesta de gran volumen o de larga duración. Esta opción expira los eventos de la ubicación del punto de comprobación, lo que acelera el tiempo de inicio del cargador automático. El tiempo de inicio puede llegar a los minutos por cada ejecución del cargador automático, lo que agrega un costo innecesario cuando se tiene un límite superior en la antigüedad máxima de los archivos que se almacenarán en el directorio de origen. El valor mínimo que se puede establecer para cloudFiles.maxFileAge es "14 days". Las eliminaciones de RocksDB aparecen como entradas de marcador de exclusión, por lo que debe esperar que el uso del almacenamiento aumente temporalmente a medida que los eventos expiran antes de que empiece a nivelarse.

Advertencia

cloudFiles.maxFileAge se ofrece como mecanismo de control de costes para conjuntos de datos de gran volumen. Un ajuste demasiado agresivo de cloudFiles.maxFileAge puede causar problemas en la calidad de los datos, como la ingesta duplicada o la falta de archivos. Por lo tanto, Databricks recomienda una configuración prudente para cloudFiles.maxFileAge, como 90 días, que es similar a lo que recomiendan soluciones comparables de ingesta de datos.

Intentar optimizar la opción cloudFiles.maxFileAge puede hacer que Auto Loader ignore los archivos sin procesar o que los archivos ya procesados expiren y, a continuación, se vuelvan a procesar, lo que provocará datos duplicados. A continuación se muestran algunos aspectos que deben tenerse en cuenta al elegir un cloudFiles.maxFileAge:

Si la secuencia se reinicia después de mucho tiempo, se omiten los eventos de notificación de archivos que se extrajeron de la cola y que son anteriores a cloudFiles.maxFileAge. Del mismo modo, si usa la lista de directorios, se omiten los archivos que pueden haber aparecido durante el tiempo de inactividad y que son anteriores a cloudFiles.maxFileAge.
Si usa el modo de lista de directorios y usa cloudFiles.maxFileAge, por ejemplo, establecido en "1 month", se detiene la secuencia y se reinicia con cloudFiles.maxFileAge establecido en "2 months"; se vuelven a procesar los archivos que son más antiguos que 1 mes, pero más recientes que 2 meses.

Si establece esta opción la primera vez que inicie el flujo, no ingerirá datos anteriores a cloudFiles.maxFileAge, por lo tanto, si quiere ingerir datos antiguos no debería establecer esta opción al iniciar su flujo por primera vez. Sin embargo, debe establecer esta opción en ejecuciones posteriores.

Desencadenar reposición normal mediante cloudFiles.backfillInterval

El cargador automático puede desencadenar reposiciones asincrónicas en un intervalo determinado, por ejemplo, un día para que la reposición se realice una vez al día o una semana para que la reposición se realice una vez a la semana. Los sistemas de notificación de eventos de archivo no garantizan al 100 % la entrega de todos los archivos que se han cargado y no proporcionan un Acuerdo de Nivel de Servicio estricto sobre la latencia de los eventos de archivo. Databricks recomienda desencadenar reposiciones periódicas con el cargador automático mediante la opción cloudFiles.backfillInterval para garantizar que todos los archivos se detectan dentro de un Acuerdo de Nivel de Servicio determinado si la integridad de los datos es un requisito. Desencadenar reposiciones periódicas no provoca duplicados.

Compartir a través de