Ingerir dados com Auto Loader

Quando novos arquivos de dados chegam continuamente ao armazenamento em nuvem, você precisa de uma maneira eficiente de processá-los sem rastrear manualmente quais arquivos foram ingeridos. O Carregador Automático resolve esse desafio detectando e ingerindo automaticamente novos arquivos conforme eles aparecem, fornecendo garantias exatamente uma vez sem exigir que você gerencie o estado ou os pontos de verificação manualmente.

Entender como o Carregador Automático funciona

O Carregador Automático é uma fonte de Streaming Estruturado que monitora um local de armazenamento em nuvem e processa incrementalmente novos arquivos. Em vez de verificar diretórios inteiros repetidamente, o Carregador Automático mantém o estado sobre quais arquivos ele já processou, garantindo que cada arquivo seja ingerido exatamente uma vez.

Quando você inicia um fluxo do Carregador Automático, ele pode processar arquivos existentes no diretório e, em seguida, observar continuamente os recém-chegados. O fluxo armazena informações de progresso em um local de ponto de verificação, permitindo que retome exatamente de onde parou se for interrompido.

O Carregador Automático detecta novos arquivos usando um dos dois modos:

• Modo de listagem de diretório: o Carregador Automático lista periodicamente o diretório de entrada para descobrir novos arquivos. Essa abordagem não requer nenhuma configuração adicional além do acesso ao armazenamento.
• Modo de notificação de arquivo: o Carregador Automático usa serviços de notificação na nuvem para receber eventos quando novos arquivos chegam. Esse modo é mais eficiente para cargas de trabalho em grande escala porque evita listagens de diretório repetidas.

Para a maioria das cargas de trabalho de produção, o Databricks recomenda o modo de notificação de arquivo com eventos de arquivo ativados na sua localização externa no Catálogo do Unity. Com eventos de arquivo, o Carregador Automático recebe notificações diretamente quando os arquivos chegam, o que reduz a latência e os custos da API de nuvem.

Configurar o Carregador Automático para ingestão de arquivo

Para ingerir dados com o Auto Loader, você usa o formato cloudFiles com spark.readStream. O exemplo a seguir lê arquivos JSON do Azure Data Lake Storage e os grava em uma tabela do Catálogo do 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"))

A cloudFiles.format opção especifica o formato do arquivo de origem. O Carregador Automático dá suporte a arquivos JSON, CSV, Parquet, Avro, ORC, XML, texto e binários. O cloudFiles.schemaLocation armazena o esquema inferido, que permite que o Carregador Automático acompanhe as alterações de esquema ao longo do tempo. Embora o local do esquema e o local do ponto de verificação possam ser os mesmos, mantê-los separados permite redefinir um fluxo sem perder o esquema inferido.

A trigger(availableNow=True) configuração processa todos os arquivos disponíveis no momento e, em seguida, é interrompida. Esse padrão funciona bem para trabalhos em lotes agendados. Para processamento contínuo, você pode omitir o gatilho ou usar trigger(processingTime="1 minute") para processar arquivos em intervalos.

Usar a sintaxe SQL com read_files

O Carregador Automático também dá suporte à sintaxe SQL por meio da read_files função com valor de tabela. Essa abordagem é útil quando você prefere SQL ou ao trabalhar em notebooks 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 em pipelines do Lakeflow, os recursos do Auto Loader são habilitados automaticamente. Use a palavra-chave STREAM para habilitar 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'
)

Nesse contexto, o pipeline gerencia de forma automática o ponto de verificação e a evolução do esquema.

Gerenciar a inferência e a evolução do esquema

O Carregador Automático pode detectar automaticamente o esquema dos arquivos de origem, eliminando a necessidade de definir esquemas manualmente. Quando a inferência de esquema está habilitada, o Carregador Automático amostra arquivos para determinar nomes de coluna e tipos de dados.

Para arquivos JSON e CSV, o Carregador Automático infere todas as colunas como cadeias de caracteres por padrão. Essa abordagem conservadora evita incompatibilidades de tipo que podem causar perda de dados. Para habilitar a detecção automática de tipo, defina a opção cloudFiles.inferColumnTypes :

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

Quando os dados de origem introduzem novas colunas, o Carregador Automático detecta a alteração e pode evoluir o esquema de destino automaticamente. A cloudFiles.schemaEvolutionMode opção controla esse comportamento:

A coluna de dados resgatada (_rescued_data) captura todos os dados que não correspondem ao esquema esperado, incluindo incompatibilidades de tipo e colunas inesperadas. Esse recurso impede a perda de dados quando os dados de origem não estão em conformidade com as expectativas. O Carregador Automático 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 resgatada, use a opção rescuedDataColumn :

.option("rescuedDataColumn", "my_rescued_data")

Se você souber que determinadas colunas devem ter tipos específicos, use dicas de esquema para substituir os tipos inferidos:

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

Monitorar e otimizar fluxos do Carregador Automático

Para cargas de trabalho de produção, o monitoramento ajuda você a acompanhar o progresso da ingestão e a identificar problemas. O Carregador Automático fornece uma função SQL para consultar o estado dos arquivos descobertos:

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

Essa consulta retorna metadados sobre arquivos descobertos pelo Carregador Automático, incluindo o status de processamento. Você pode usar essas informações para verificar se os arquivos estão sendo ingeridos conforme o esperado.

O Auto Loader relata métricas por meio do painel de progresso do Structured Streaming, incluindo:

• numFilesOutstanding: arquivos descobertos, mas ainda não processados
• numBytesOutstanding: total de bytes aguardando para serem processados

Para cargas de trabalho de alto volume, controle a taxa de processamento usando cloudFiles.maxFilesPerTrigger ou cloudFiles.maxBytesPerTrigger. Essas configurações impedem que qualquer microlote único se torne muito grande:

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

Para acessar metadados no nível do arquivo em seus dados ingeridos, selecione a _metadata coluna. Esta coluna oculta contém informações sobre cada arquivo de origem, incluindo file_path, file_namee file_sizefile_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 aos pipelines do Lakeflow, em que lida automaticamente com o gerenciamento de esquemas e pontos de verificação. Na próxima unidade, você aprenderá a usar o Carregador Automático dentro de definições declarativas de pipeline para fluxos de trabalho de ingestão de dados de ponta a ponta.