Ingesta de datos con cargador automático

Cuando los nuevos archivos de datos llegan continuamente en el almacenamiento en la nube, necesita una manera eficaz de procesarlos sin realizar un seguimiento manual de los archivos que se han ingerido. Auto Loader resuelve este desafío mediante la detección y ingesta automática de nuevos archivos tal como aparecen, proporcionando garantías exactamente una vez sin necesidad de administrar manualmente el estado o los puntos de control.

Comprender cómo funciona Auto Loader

Auto Loader es un origen de Structured Streaming que supervisa una ubicación de almacenamiento en la nube y procesa incrementalmente los nuevos archivos. En lugar de examinar directorios completos repetidamente, Auto Loader mantiene el estado sobre los archivos que ya ha procesado, lo que garantiza que cada archivo se ingiere exactamente una vez.

Al iniciar una secuencia de Auto Loader, puede procesar los archivos existentes en el directorio y luego continuar observando continuamente las nuevas llegadas. El flujo almacena información de progreso en una ubicación de punto de control, lo que le permite reanudar exactamente desde el punto en que se detuvo si se interrumpe.

Auto Loader detecta nuevos archivos mediante uno de los dos modos:

• Modo de lista de directorios: el cargador automático muestra periódicamente el directorio de entrada para detectar nuevos archivos. Este enfoque no requiere ninguna configuración adicional más allá del acceso al almacenamiento.
• Modo de notificación de archivo: El cargador automático usa servicios de notificaciones en la nube para recibir eventos cuando llegan nuevos archivos. Este modo es más eficaz para cargas de trabajo a gran escala, ya que evita listas de directorios repetidas.

Para la mayoría de las cargas de trabajo de producción, Databricks recomienda el modo de notificación de archivos con eventos de archivos habilitados en su ubicación externa en Unity Catalog. Con los eventos de archivo, Auto Loader recibe notificaciones directamente cuando llegan los archivos, lo que reduce la latencia y los costos de la API en la nube.

Configuración del cargador automático para la ingesta de archivos

Para ingerir datos con Auto Loader, utilice el formato cloudFiles con spark.readStream. En el ejemplo siguiente se leen los archivos JSON de Azure Data Lake Storage y se escriben en una tabla de Catálogo de Unity:

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

La cloudFiles.format opción especifica el formato de archivo de origen. Auto Loader admite archivos JSON, CSV, Parquet, Avro, ORC, XML, text y binarios. cloudFiles.schemaLocation almacena el esquema inferido, que permite al cargador automático realizar un seguimiento de los cambios de esquema a lo largo del tiempo. Aunque la ubicación del esquema y la ubicación del punto de control pueden ser las mismas, mantenerlas separadas le permite restablecer una secuencia sin perder el esquema inferido.

La trigger(availableNow=True) configuración procesa todos los archivos disponibles actualmente y, a continuación, se detiene. Este patrón funciona bien para los trabajos por lotes programados. Para el procesamiento continuo, puede omitir el desencadenador o usar trigger(processingTime="1 minute") para procesar archivos en intervalos.

Uso de la sintaxis SQL con read_files

Auto Loader también admite la sintaxis SQL a través de la read_files función con valores de tabla. Este enfoque es útil cuando se prefiere SQL o cuando se trabaja en cuadernos de SQL:

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Cuando se usa read_files con tablas de streaming en canalizaciones de Lakeflow, las funcionalidades de Auto Loader se habilitan automáticamente. Use la palabra clave STREAM para habilitar el procesamiento incremental:

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

En este contexto, el pipeline gestiona automáticamente los puntos de control y la evolución del esquema.

Manejar la inferencia de esquemas y la evolución

Auto Loader puede detectar automáticamente el esquema de los archivos de origen, lo que elimina la necesidad de definir esquemas manualmente. Cuando la inferencia de esquema está habilitada, el cargador automático muestra archivos para determinar los nombres de columna y los tipos de datos.

En el caso de los archivos JSON y CSV, el cargador automático deduce todas las columnas como cadenas de forma predeterminada. Este enfoque conservador evita desajustes de tipos que podrían causar pérdida de datos. Para habilitar la detección automática de tipos, establezca la cloudFiles.inferColumnTypes opción :

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

Cuando los datos de origen introducen nuevas columnas, Auto Loader detecta el cambio y puede evolucionar automáticamente el esquema de destino. La cloudFiles.schemaEvolutionMode opción controla este comportamiento:

La columna de datos rescate (_rescued_data) captura los datos que no coinciden con el esquema esperado, incluidos los errores de coincidencia de tipos y las columnas inesperadas. Esta característica evita la pérdida de datos cuando los datos de origen no se ajustan a las expectativas. Auto Loader agrega automáticamente esta columna al deducir el esquema.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

Para cambiar el nombre de la columna de datos rescatados, use la opción rescuedDataColumn.

.option("rescuedDataColumn", "my_rescued_data")

Si sabe que ciertas columnas deben tener tipos específicos, use sugerencias de esquema para invalidar los tipos inferidos:

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

Supervisión y optimización de flujos del cargador automático

En el caso de las cargas de trabajo de producción, la supervisión le ayuda a realizar un seguimiento del progreso de la ingesta e identificar problemas. Auto Loader proporciona una función SQL para consultar el estado de los archivos detectados:

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

Esta consulta devuelve metadatos sobre los archivos detectados por Auto Loader, incluido su estado de procesamiento. Puede usar esta información para comprobar que los archivos se ingieren según lo previsto.

El cargador automático informa de las métricas a través del panel de progreso de Structured Streaming, entre los que se incluyen:

• numFilesOutstanding: archivos detectados pero aún no procesados
• numBytesOutstanding: bytes totales en espera de procesamiento

Para cargas de trabajo de gran volumen, controle la velocidad de procesamiento mediante cloudFiles.maxFilesPerTrigger o cloudFiles.maxBytesPerTrigger. Esta configuración impide que cualquier único microproceso sea demasiado grande:

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Para acceder a los metadatos de nivel de archivo en los datos ingeridos, seleccione la _metadata columna . Esta columna oculta contiene información sobre cada archivo de origen, incluidos file_path, file_name, file_sizey file_modification_time:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader se integra sin problemas con las canalizaciones de Lakeflow, donde controla los puntos de control y la administración de esquemas automáticamente. En la siguiente unidad, aprenderá a utilizar Auto Loader dentro de definiciones de canalizaciones declarativas para flujos de trabajo de ingesta de datos de extremo a extremo.