Melhores práticas do Auto Loader

Esta página descreve as melhores práticas que pode aplicar para configurar o Auto Loader para funcionar de forma fiável, económica e em escala para o seu caso de uso.

Estas boas práticas reduzem a sobrecarga operacional e evitam problemas comuns que são difíceis de diagnosticar em produção, tais como: custos desnecessários da API LIST devido a varrimentos completos de diretórios, perda silenciosa de dados devido a alterações no esquema e reinícios do pipeline causados por má configuração do ponto de verificação.

Para detalhes de configuração em produção, consulte Configurar Auto Loader para cargas de trabalho de produção. Para monitorização e observabilidade, consulte Monitorizar e observar o Auto Loader.

Escolha a estrutura de execução certa

A melhor estrutura de execução para o seu caso de utilização depende do grau de controlo de que necessita sobre o pipeline e da sobrecarga operacional que pretende gerir. Para a maioria dos utilizadores e pipelines de produção, o Auto Loader com pipelines Lakeflow é uma boa opção. No entanto, se precisares de máximo controlo e personalização, usa o Auto Loader com Structured Streaming. Para a configuração mais simples com uma experiência gerida, utilize um LakeFlow Connector gerido quando disponível.

Os pipelines do Lakeflow expandem o Structured Streaming com dimensionamento automático, verificações da qualidade dos dados, gestão da evolução do esquema e monitorização através do registo de eventos. A Databricks recomenda os pipelines Lakeflow para a maioria das cargas de trabalho de ingestão em ambiente de produção.

Escolha o horário e o tipo de gatilho certos

O melhor tipo de agendamento e de gatilho para o seu caso de uso depende dos seus requisitos de latência e dos padrões de chegada de ficheiros. Para a maioria dos casos de uso, o Databricks recomenda um gatilho de chegada de ficheiros com eventos de ficheiro ativados. Isto permite uma ingestão de baixa latência a baixo custo porque o cálculo só é executado quando chegam novos ficheiros. Os três tipos de gatilho diferem em quando e com que frequência o pipeline inicia:

• Contínuo: O oleoduto funciona sem parar. Use apenas quando uma latência inferior a um segundo for um requisito rígido, pois o cálculo contínuo custa mais. Associar a eventos de ficheiro.
• Acionador de chegada de ficheiros: O pipeline inicia-se quando novos ficheiros são colocados na localização de origem. Ideal para latência baixa a média ou padrões irregulares de chegada de ficheiros. Requer que os eventos do ficheiro estejam ativados. Consulte Acionar trabalhos quando novos arquivos chegam.
• Programado: O pipeline é executado de acordo com uma programação temporal (por exemplo, a cada hora). Use quando os requisitos de latência forem flexíveis (minutos a horas). Funciona com listagem de diretórios, mas os eventos de ficheiros reduzem custos mesmo em modo agendado ao evitar varreções completas de diretórios.

Para obter detalhes sobre a utilização de Trigger.AvailableNow para agendamento por lotes, consulte Utilizar Trigger.AvailableNow e limitação da taxa.

Escolha o modo certo de descoberta de ficheiros

O Auto Loader suporta três modos de deteção de ficheiros, com diferentes compensações ao nível da complexidade de configuração, da escalabilidade e do custo.

Os eventos de ficheiros consolidam os recursos de armazenamento na cloud usando uma subscrição e fila por localização externa em vez de uma por fluxo. A diferença de desempenho é significativa em grande escala: a listagem de diretórios tem de escanear todo o diretório fonte em cada disparador, pelo que o tempo de ingestão aumenta com o tamanho do diretório. Os eventos de ficheiro entregam notificações de novos ficheiros diretamente, por isso o tempo de ingestão mantém-se baixo independentemente de quantos objetos estejam no diretório.

Ativar eventos de ficheiro

Os eventos de ficheiro requerem uma concessão única de permissões na cloud e uma localização externa configurada para usar o serviço de eventos de ficheiros geridos. Uma vez configurado, todos os fluxos do Auto Loader que leem a partir dessa localização externa podem usar eventos de ficheiro sem configuração adicional.

1. Conceda as permissões necessárias ao fornecedor de cloud. Os requisitos variam consoante o fornecedor de cloud. Veja Definir eventos de ficheiro para uma localização externa.

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

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

   Para os passos completos de configuração, veja Migrar para o Auto Loader com eventos de ficheiros.

Quando não pode usar eventos de ficheiro

Pode não conseguir usar eventos de ficheiro quando:

• A localização externa não está configurada com eventos de ficheiros.
• As políticas de segurança da organização não permitem ativar eventos de ficheiros numa localização externa partilhada.

Nestes casos, utilize o modo clássico de notificação de ficheiros ou o modo de listagem de diretórios. Para uma comparação completa dos modos de deteção de ficheiros, veja Compare os modos de deteção de ficheiros do Auto Loader.

Gerir evolução de esquemas

O Auto Loader infere o esquema automaticamente, mas a forma como configuras a evolução do esquema afeta a completude dos dados e a estabilidade do pipeline. Use a tabela seguinte para escolher uma estratégia.

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

Use dicas de esquema para tipos de campos conhecidos

Use a cloudFiles.schemaHints opção de impor tipos para campos que já conhece, permitindo ainda a inferência de esquemas para outros campos.

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

Utilize a expansão de tipo para alterações de tipo compatíveis

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

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

Importar como tipo Variant para esquemas imprevisíveis

Quando os seus dados não correspondem a nenhum esquema específico, ou o esquema muda continuamente, ingera os dados como um Variant tipo.

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

Variant fornece schema-on-read no momento da consulta, mas é menos eficiente do que consultar colunas estruturadas. Para a mecânica completa da inferência e evolução de esquemas, veja Configurar inferência e evolução de esquemas no Auto Loader.

Lidar com dados incorretos e com a qualidade dos dados

As seguintes práticas ajudam-no a detetar, capturar e isolar dados errados antes que se propaguem para camadas posteriores.

Habilitar _rescued_data e _corrupt_record

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

• _rescued_data captura campos que não correspondem ao esquema atual. É adicionado automaticamente pelo Auto Loader.
• _corrupt_record regista linhas que não podem ser de todo analisadas. Ative-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 recomenda columnNameOfCorruptRecord em vez de badRecordsPath para evitar potenciais condições de corrida que podem não detetar registos corrompidos.

Utilizar as expectativas dos oleodutos de fluxo de lago para monitorização

Defina expectativas nos pipelines do Lakeflow para verificar que _rescued_data e _corrupt_record se encontram NULL em condições normais. Valores não-NULL sinalizam desvio do 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 as linhas que contenham dados não analisáveis num sumidouro dedicado para investigação. Isto impede que dados corrompidos se propaguem para camadas a jusante.

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 do ficheiro de origem

Inclua a coluna _metadata nas consultas de ingestão do Auto Loader. No mínimo, capture file_path e file_modification_time. Isto permite-lhe rastrear problemas de dados até aos ficheiros de origem específicos e fazer junção com cloud_files_state() ao longo de todo o ciclo de vida do ficheiro.

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

Para mais detalhes, consulte a coluna de metadados do ficheiro.

Otimizar custos e desempenho

As seguintes práticas reduzem os três principais fatores de custo do Auto Loader: chamadas de API na cloud LIST , computação ociosa e crescimento do armazenamento a longo prazo.

• Usar eventos de ficheiro para minimizar LIST Custos da API: Os eventos de ficheiros proporcionam descoberta incremental de ficheiros, eliminando a necessidade de listagens completas de diretórios em cada execução. Esta é a otimização de custos mais impactante para o Auto Loader.

• Utilize acionadores de chegada de ficheiros para processamento orientado a eventos: Os acionadores de chegada de ficheiros só iniciam o seu pipeline quando chegam novos ficheiros, pelo que não paga por capacidade de computação inativa. Consulte Acionar trabalhos quando novos arquivos chegam.

• Arquivar ficheiros processados com cloudFiles.cleanSource: Usar cloudFiles.cleanSource para eliminar ou mover automaticamente ficheiros processados. Isto reduz tanto os custos de armazenamento como os custos de listagem de diretórios para fluxos de longa duração. Para detalhes completos, consulte Arquivamento de ficheiros no diretório de origem para reduzir custos.

  • Utilize o modo delete para remover ficheiros após a ingestão.
  • Utilize o modo move para arquivar ficheiros noutro local para efeitos de conformidade ou auditoria.

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

  Warning

  Não ative cloudFiles.cleanSource se múltiplos fluxos do Auto Loader ou outros clientes lerem do mesmo diretório de origem.

• Aproveite as melhorias de desempenho: Atualize para o mais recente Databricks Runtime ou utilize computação serverless para beneficiar das recentes melhorias de desempenho do Auto Loader.

Gestão de pontos de controlo

O ponto de controlo armazena o progresso do fluxo e o estado do ficheiro. Configurar mal ou perder o ponto de controlo requer um reinício completo, por isso trate-o como infraestrutura crítica.

• Nunca aplique políticas de ciclo de vida de objetos na nuvem às localizações dos pontos de controlo. Se os ficheiros checkpoint forem apagados, o estado do stream fica corrompido e tens de reiniciar do zero.
• Utilize pontos de verificação separados para cada fluxo e diretório de origem.
• Considere cloudFiles.maxFileAge para fluxos de longa duração e elevado volume, para limitar o crescimento do estado. Utilize uma definição conservadora (recomenda-se um mínimo de 90 dias). Definir este valor de forma demasiado agressiva corre o risco de reprocessar ficheiros que o Auto Loader já ingeriu se estes ficarem fora da janela.

Para detalhes completos, consulte rastreamento de eventos de ficheiros.

Utilize volumes para uma deteção ideal de ficheiros com eventos de ficheiro

Para melhorar o desempenho com eventos de ficheiros, crie um volume externo para cada caminho ou subdiretório de onde o Auto Loader carrega. Fornecer caminhos de volume (por exemplo, /Volumes/catalog/schema/volume) ao Auto Loader em vez de caminhos cloud (por exemplo, s3://bucket/path). Isto otimiza a descoberta de ficheiros através de um padrão de acesso otimizado aos dados.

Para mais boas práticas para eventos de ficheiros, consulte Boas práticas para Auto Loader com eventos de ficheiro.