Práticas recomendadas do Carregador Automático

Esta página descreve as práticas recomendadas que você pode aplicar para configurar o Carregador Automático para ser executado de forma confiável, econômica e em escala para seu caso de uso.

Essas práticas recomendadas reduzem a sobrecarga operacional e evitam problemas comuns que são difíceis de diagnosticar em produção, como: custos desnecessários de API causados por varreduras completas de diretório LIST, perda silenciosa de dados devido à deriva de esquema e reinicializações do pipeline causadas por configuração incorreta do ponto de verificação.

Para obter detalhes de configuração de produção, consulte Configurar o Carregador Automático para cargas de trabalho de produção. Para monitoramento e observabilidade, consulte Monitorar e observar o Carregador Automático.

Escolha a estrutura de execução correta

A melhor estrutura de execução para seu caso de uso depende de quanto controle você precisa sobre o pipeline e de quanta sobrecarga operacional você deseja gerenciar. Para a maioria dos usuários e pipelines de produção, o Auto Loader com pipelines do Lakeflow é uma boa escolha. No entanto, se você precisar de controle máximo e personalização, use o Carregador Automático com Streaming Estruturado. Para uma configuração mais simples e uma experiência totalmente gerenciada, use um conector gerenciado do LakeFlow quando disponível.

Os pipelines do Lakeflow estendem o Streaming Estruturado com dimensionamento automático, verificações de qualidade de dados, tratamento de evolução do esquema e monitoramento por meio do log de eventos. O Databricks recomenda pipelines Lakeflow para a maioria das cargas de trabalho de ingestão em produção.

Escolha o agendamento e o tipo de acionador corretos

O melhor tipo de agendamento e gatilho para seu caso de uso depende dos requisitos de latência e dos padrões de chegada do arquivo. Para a maioria dos casos de uso, a Databricks recomenda um gatilho de chegada de arquivo com eventos de arquivo habilitados. Isso permite ingestão de baixa latência a baixo custo, porque os recursos computacionais só são acionados quando novos arquivos chegam. Os três tipos de gatilho diferem em quando e com que frequência o pipeline é iniciado:

• Contínuo: o pipeline é executado sem parar. Use somente quando a latência de sub-segundo for um requisito difícil, já que a computação contínua custa mais. Emparelhe com eventos de arquivo.
• Gatilho de chegada de arquivos: o pipeline é iniciado quando novos arquivos chegam ao local de origem. Melhor para latência baixa a média ou padrões irregulares de chegada de arquivo. Requer que os eventos de arquivo sejam habilitados. Veja Acione trabalhos quando novos arquivos chegam.
• Agendado: o pipeline é executado de acordo com uma programação baseada em tempo (por exemplo, a cada hora). Use-o quando os requisitos de latência forem flexíveis (de minutos a horas). Funciona com listagem de diretórios, mas os eventos de arquivo reduzem custos mesmo no modo agendado ao evitar verificações completas de diretório.

Para obter detalhes sobre como usar Trigger.AvailableNow para agendamento em lote, consulte Como usar Trigger.AvailableNow e limitação de taxa.

Escolher o modo de descoberta de arquivo correto

O Carregador Automático dá suporte a três modos de descoberta de arquivos com diferentes compensações na complexidade da instalação, escalabilidade e custo.

Os eventos de arquivo consolidam os recursos de armazenamento em nuvem usando uma assinatura e uma fila por local externo, em vez de uma por fluxo. A diferença de desempenho é significativa em escala: a listagem de diretórios deve examinar todo o diretório de origem em cada gatilho, de modo que o tempo de ingestão cresça com o tamanho do diretório. Os eventos de arquivo fornecem novas notificações de arquivo diretamente, portanto, o tempo de ingestão permanece baixo, independentemente de quantos objetos estão no diretório.

Habilitar eventos de arquivo

Os eventos de arquivo exigem uma concessão de permissões de nuvem única e um local externo configurado para usar o serviço de eventos de arquivo gerenciado. Depois de configurados, todos os fluxos do Auto Loader que leem desse local externo podem usar eventos de arquivo sem configuração adicional.

1. Conceda as permissões de nuvem necessárias no lado do provedor de nuvem. Os requisitos variam de acordo com o provedor de nuvem. Veja Configurar eventos de arquivo para um local externo.

2. Defina cloudFiles.useManagedFileEvents para true na sua consulta do Auto Loader.

   df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.useManagedFileEvents", "true")
     .load("/path/to/data/dir"))
   

   Para obter as etapas completas de instalação, consulte Migrar para o Carregador Automático com eventos de arquivo.

Quando você não pode usar eventos de arquivo

Talvez você não possa usar eventos de arquivo quando:

• O local externo não está configurado com eventos de arquivo.
• As políticas de segurança da organização não permitem habilitar eventos de arquivo em um local externo compartilhado.

Nesses casos, use o modo de notificação de arquivo clássico ou o modo de listagem de diretórios. Para obter uma comparação completa dos modos de detecção de arquivos, consulte Comparar modos de detecção de arquivo do Carregador Automático.

Gerenciar a evolução do esquema

O Carregador Automático infere o esquema automaticamente, mas como você configura a evolução do esquema afeta a integridade dos dados e a estabilidade do pipeline. Use a tabela a seguir para escolher uma estratégia.

Depois de escolher uma estratégia, aplique as seguintes práticas para ajustar como a evolução do esquema se comporta.

Usar dicas de esquema para tipos de campo conhecidos

Use a opção cloudFiles.schemaHints para impor tipos para campos que você conhece com antecedência, ao mesmo tempo em que permite inferência de esquema para outros campos.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "id long, amount double")
  .load("/path/to/data/dir"))

Usar ampliação de tipo para alterações de tipo compatíveis

O addNewColumnsWithTypeWidening modo de evolução do esquema amplia automaticamente tipos compatíveis (por exemplo, int para long) em vez de rotear dados para a _rescued_data coluna. Isso evita a necessidade de trabalhos pós-processamento para lidar com promoções de tipo simples.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load("/path/to/data/dir"))

Ingerir como tipoVariantpara esquemas imprevisíveis

Quando seus dados não se ajustarem a nenhum esquema específico, ou se o esquema mudar continuamente, ingira os dados como o tipo Variant.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "data")
  .load("/path/to/data/dir"))

Variantfornece esquema na leitura no momento da consulta, mas é menos eficiente do que consultar colunas estruturadas. Para obter a mecânica completa da inferência e da evolução do esquema, consulte Configurar a inferência e a evolução do esquema no Carregador Automático.

Lidar com dados inválidos e com a qualidade dos dados

As práticas a seguir ajudam você a detectar, capturar e isolar dados incorretos antes que eles se propaguem para camadas downstream.

Habilitar _rescued_data e _corrupt_record

O Auto Loader disponibiliza duas colunas para registrar dados que não podem ser analisados corretamente.

• _rescued_data captura campos que não correspondem ao esquema atual. Ele é adicionado automaticamente pelo Carregador Automático.
• _corrupt_record captura linhas que não podem ser analisadas. Habilite-o usando columnNameOfCorruptRecord:

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "_corrupt_record string")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .load("/path/to/data/dir"))

A Databricks recomendacolumnNameOfCorruptRecordem vez debadRecordsPathpara evitar possíveis condições de corrida que podem deixar de detectar registros corrompidos.

Use as expectativas dos pipelines do Lakeflow para monitoramento

Defina as expectativas dos pipelines do Lakeflow para verificar se _rescued_data e _corrupt_record estão NULL em condições normais. Valores não NULL sinalizam descompasso de esquema ou corrupção de dados.

import dlt

@dlt.table
@dlt.expect("no rescued data", "_rescued_data IS NULL")
@dlt.expect("no corrupt records", "_corrupt_record IS NULL")
def bronze_table():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", "_corrupt_record string")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .load("/path/to/data/dir"))

Isolar dados corrompidos

Isole linhas com dados não analisáveis em um coletor dedicado para investigação. Isso impede que dados corrompidos se propaguem para camadas downstream.

import dlt

@dlt.table
def corrupt_records_sink():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NOT NULL")

@dlt.view
def clean_table():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NULL")

Anotar dados com metadados de arquivo de origem

Inclua a coluna_metadatanas consultas de ingestão do Auto Loader. No mínimo, capture file_path e file_modification_time. Isso permite rastrear problemas de dados até arquivos de origem específicos e fazer junção comcloud_files_state()para todo o ciclo de vida do arquivo.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/path/to/data/dir")
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Para obter detalhes, consulte a coluna de metadados de arquivo.

Otimizar o custo e o desempenho

As práticas a seguir reduzem os três principais drivers de custo para o Carregador Automático: chamadas à API de nuvem LIST , computação ociosa e crescimento de armazenamento de longo prazo.

• Usar eventos de arquivo para minimizar LIST Custos da API: os eventos de arquivo fornecem a descoberta incremental de arquivos, eliminando a necessidade de listagens completas de diretório em cada execução. Essa é a otimização de custo mais impactante para o Carregador Automático.

• Use gatilhos de chegada de arquivo para processamento orientado a eventos: os gatilhos de chegada de arquivo iniciam o pipeline apenas quando novos arquivos chegam, para que você não pague por computação ociosa. Veja Acione trabalhos quando novos arquivos chegam.

• Arquivar arquivos processados com cloudFiles.cleanSource: use cloudFiles.cleanSource para excluir ou mover arquivos processados automaticamente. Isso reduz os custos de armazenamento e os custos de listagem de diretórios para fluxos de longa duração. Para obter detalhes completos, consulte Arquivos de arquivamento no diretório de origem para reduzir custos.

  • Use o delete modo para remover arquivos após a ingestão.
  • Use o modo move para arquivar arquivos em um local diferente para fins de conformidade ou auditoria.

  df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.cleanSource", "delete")
    .load("/path/to/data/dir"))
  

  Aviso

  Não habilite cloudFiles.cleanSource se vários fluxos de Auto Loader ou outros clientes lerem do mesmo diretório de origem.

• Aproveite as melhorias de desempenho: atualize para o Databricks Runtime mais recente ou use a computação sem servidor para se beneficiar de melhorias recentes de desempenho do Carregador Automático.

Gerenciamento de ponto de verificação

O ponto de verificação armazena o progresso do fluxo e o estado do arquivo. Configurar incorretamente ou perder o ponto de verificação exige uma reinicialização completa, portanto trate isso como infraestrutura crítica.

• Nunca aplique políticas de ciclo de vida de objetos de nuvem a locais de ponto de verificação. Se os arquivos de ponto de verificação forem excluídos, o estado do fluxo será corrompido e você deverá reiniciar do zero.
• Use pontos de verificação separados para cada fluxo e diretório de origem.
• Considere cloudFiles.maxFileAge para fluxos de longa duração e alto volume, a fim de limitar o crescimento do estado. Use uma configuração conservadora (mínimo de 90 dias recomendado). Definir esse valor de forma muito agressiva corre o risco de reprocessar arquivos que o Carregador Automático já ingeriu se eles estiverem fora da janela.

Para obter detalhes completos, consulte o acompanhamento de eventos de arquivo.

Use volumes para descoberta ideal de arquivos com eventos de arquivo

Para melhorar o desempenho com eventos de arquivo, crie um volume externo para cada caminho ou subdiretório do qual o Carregador Automático carrega. Forneça caminhos de volume (por exemplo, /Volumes/catalog/schema/volume) para o Carregador Automático em vez de caminhos de nuvem (por exemplo, s3://bucket/path). Isso otimiza a descoberta de arquivos por meio de um padrão de acesso a dados otimizado.

Para obter mais práticas recomendadas de eventos de arquivo, consulte As práticas recomendadas para o Carregador Automático com eventos de arquivo.