Ingestão de dados com o Auto Loader

Quando novos ficheiros de dados chegam continuamente ao armazenamento na cloud, é necessário uma forma eficiente de os processar sem rastrear manualmente quais os ficheiros que foram ingeridos. O Auto Loader resolve este desafio ao detetar e ingerir automaticamente novos ficheiros à medida que surgem, garantindo execução exata de uma vez, sem necessidade de gerir manualmente o estado ou os checkpoints.

Compreender como funciona o Auto Loader

O Auto Loader é uma fonte de Streaming Estruturado que monitoriza uma localização de armazenamento na cloud e processa novos ficheiros de forma incremental. Em vez de analisar diretórios inteiros repetidamente, o Auto Loader mantém o estado sobre quais ficheiros já processou, garantindo que cada ficheiro é ingerido exatamente uma vez.

Quando inicias um stream do Auto Loader, ele pode processar ficheiros existentes no diretório e depois vigiar continuamente as novas chegadas. O fluxo armazena informação de progresso num ponto de controlo, o que lhe permite retomar exatamente onde parou se for interrompido.

O Auto Loader deteta ficheiros novos usando um de dois modos:

• Modo de listagem de diretórios: O Auto Loader lista periodicamente o diretório de entrada para descobrir novos ficheiros. Esta abordagem não requer configuração adicional além do acesso ao armazenamento.
• Modo de notificação de ficheiros: O Auto Loader utiliza serviços de notificação na cloud para receber eventos quando chegam novos ficheiros. Este modo é mais eficiente para cargas de trabalho de grande escala porque evita listagens repetidas de diretórios.

Para a maioria das cargas de trabalho de produção, o Databricks recomenda o modo de notificação de ficheiros com eventos de ficheiro ativados na sua localização externa no Unity Catalog. Com eventos de ficheiros, o Auto Loader recebe notificações diretamente quando os ficheiros chegam, o que reduz a latência e os custos da API cloud.

Configurar o Auto Loader para a ingestão de ficheiros

Para ingerir dados com o Auto Loader, utiliza-se o cloudFiles formato com spark.readStream. O exemplo seguinte lê ficheiros JSON do Azure Data Lake Storage e grava-os numa tabela do Unity Catalog:

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"))

A cloudFiles.format opção especifica o formato do ficheiro de origem. O Auto Loader suporta ficheiros JSON, CSV, Parquet, Avro, ORC, XML, texto e binários. Armazena cloudFiles.schemaLocation o esquema inferido, que permite ao Auto Loader acompanhar as alterações do esquema ao longo do tempo. Embora a localização do esquema e do ponto de controlo possam ser iguais, mantê-los separados permite reiniciar um fluxo sem perder o esquema inferido.

A trigger(availableNow=True) definição processa todos os ficheiros disponíveis e depois para. Este padrão funciona bem para trabalhos em lote agendados. Para processamento contínuo, podes omitir o trigger ou usar trigger(processingTime="1 minute") para processar ficheiros em intervalos.

Utilize a sintaxe SQL com read_files

O Auto Loader também suporta sintaxe SQL através da read_files função de valores de tabela. Esta abordagem é útil quando prefere SQL ou quando trabalha em cadernos SQL:

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

Ao usar read_files com tabelas de streaming nos pipelines Lakeflow, as capacidades do Auto Loader são ativadas automaticamente. Use a STREAM palavra-chave para permitir o processamento 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'
)

Neste contexto, o pipeline gere automaticamente o checkpointing e a evolução do esquema.

Gerir a inferência e evolução de esquemas

O Auto Loader pode detetar automaticamente o esquema dos seus ficheiros fonte, eliminando a necessidade de definir esquemas manualmente. Quando a inferência de esquema está ativada, o Auto Loader amostra ficheiros para determinar nomes de colunas e tipos de dados.

Para ficheiros JSON e CSV, o Auto Loader infere todas as colunas como strings por defeito. Esta abordagem conservadora evita incompatibilidades de tipos que possam causar perda de dados. Para ativar a deteção automática de tipos, defina a cloudFiles.inferColumnTypes opção:

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

Quando os seus dados de origem introduzem novas colunas, o Auto Loader deteta a alteração e pode evoluir automaticamente o esquema de destino. A cloudFiles.schemaEvolutionMode opção controla este comportamento:

A coluna de dados resgatados (_rescued_data) capta quaisquer dados que não correspondam ao esquema esperado, incluindo incompatibilidades de tipo e colunas inesperadas. Esta funcionalidade previne a perda de dados quando os dados de origem não cumprem as expectativas. O Auto Loader adiciona automaticamente esta coluna ao inferir o esquema.

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

Para renomear a coluna de dados resgatados, use a rescuedDataColumn opção:

.option("rescuedDataColumn", "my_rescued_data")

Se souber que certas colunas devem ter tipos específicos, use dicas de esquema para sobrepor os tipos inferidos:

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

Monitorizar e otimizar os fluxos do Auto Loader

Para cargas de trabalho em produção, a monitorização ajuda-te a acompanhar o progresso da ingestão e a identificar problemas. O Auto Loader fornece uma função SQL para consultar o estado dos ficheiros descobertos:

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

Esta consulta devolve metadados sobre ficheiros que o Auto Loader descobriu, incluindo o seu estado de processamento. Pode usar esta informação para verificar se os ficheiros estão a ser ingeridos conforme esperado.

O Auto Loader reporta métricas através do painel de progresso do Structured Streaming, incluindo:

• numFilesOutstanding: Ficheiros descobertos mas ainda não processados
• numBytesOutstanding: Bytes totais à espera de serem processados

Para cargas de trabalho de alto volume, controla a taxa de processamento usando cloudFiles.maxFilesPerTrigger ou cloudFiles.maxBytesPerTrigger. Estas definições impedem que qualquer micro-lote se torne demasiado grande:

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

Para aceder aos metadados ao nível do ficheiro nos seus dados ingeridos, selecione a _metadata coluna. Esta coluna oculta contém informação sobre cada ficheiro fonte, incluindo file_path, file_name, file_size, e file_modification_time:

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

O Auto Loader integra-se perfeitamente com os pipelines Lakeflow, onde gere automaticamente checkpoints e gestão de esquemas. Na unidade seguinte, aprende a usar o Auto Loader dentro de definições declarativas de pipelines para fluxos de trabalho de ingestão de dados de ponta a ponta.