Pular para o conteúdo principal

Auto Loader

validado

Use Auto Loader quando arquivos chegam ao longo do tempo e o Databricks deve descobri-los com checkpoints. ContractForge suporta execução finite available-now e registra evidências no stream e nos child batches.

Quando usar​

Bounded streaming

Ingestão available-now​

Execute o Auto Loader até que os arquivos atualmente disponíveis sejam processados, então pare e retorne um resultado de stream.

Fora do escopo

Aplicativos de streaming contínuo​

Aplicativos de streaming de longa duração devem ser implementados como trabalhos de streaming dedicados fora do ContractForge.

Evidência operacional

Child runs​

Cada microbatch executa um ingest_plan interno com um child run ID vinculado à stream run.

Métricas de fluxo

Métricas agregadas​

ctrl_ingestion_streams armazena a contagem de lotes processados ​​e o total de linhas lidas/gravadas/em quarentena.

Requisitos de runtime​

ExigênciaDetalhes
Databricks Auto LoaderO runtime deve suportar cloudFiles e structured streaming em modo available-now.
Source accessO cluster ou ambiente serverless deve ler o landing path por Volume, External Location ou credenciais cloud.
Acesso ao ponto de verificaçãoO runtime deve ler e gravar o checkpoint location entre tentativas e reinícios de job.
Localização do esquemaO schema tracking path deve ser durável e específico da tabela.

Exemplo JSON available-now​

source:
  type: connector
  connector: autoloader
  format: json
  path: /Volumes/main/landing/orders_json
  read:
    checkpoint_location: /Volumes/main/ops/checkpoints/orders_json
    schema_location: /Volumes/main/ops/schemas/orders_json
    available_now: true
    schema: "order_id STRING, updated_at TIMESTAMP, payload STRUCT<amount:DOUBLE,status:STRING>"
    cloudFiles.includeExistingFiles: true
    cloudFiles.maxFilesPerTrigger: 100

target:
  catalog: main
  schema: bronze_sales
  table: b_orders_autoloader

layer: bronze
mode: append
quality_rules:
  not_null: [order_id]

Exemplo de CSV disponível agora​

source:
  type: connector
  connector: autoloader
  format: csv
  path: /Volumes/main/landing/cdc_csv
  read:
    checkpoint_location: /Volumes/main/ops/checkpoints/cdc_csv
    schema_location: /Volumes/main/ops/schemas/cdc_csv
    available_now: true
    schema: "submission_date DATE, state STRING, tot_cases BIGINT"
    cloudFiles.includeExistingFiles: true
    cloudFiles.schemaHints: "submission_date DATE, state STRING, tot_cases BIGINT"
  options:
    header: true
    delimiter: ","

target:
  catalog: main
  schema: bronze_health
  table: b_cdc_autoloader

layer: bronze
mode: append

Ciclo de vida de execução​

  1. Iniciar execução do stream. ContractForge grava um registro pai em ctrl_ingestion_streams.
  2. Abra o Auto Loader readStream. O resolvedor de origem cria um leitor cloudFiles a partir de opções de contrato.
  3. Processar microlotes. Cada lote chama ingest_plan internamente com source=batch_df.
  4. Métricas agregadas. As métricas de resultados em lote são normalizadas e gravadas na tabela de controle de fluxo.
  5. Concluir ou falhar. O status e os erros do fluxo persistem antes que o chamador receba sucesso ou uma exceção.

Locais de checkpoints e esquemas​

Use locais estáveis ​​que sobrevivam às reinicializações do trabalho. Evite caminhos de espaço de trabalho efêmeros para checkpoints de produção.

LocalizaçãoPropósitoRecomendação
checkpoint_locationRastreia arquivos processados ​​e estado de fluxo.Use um volume controlado ou um caminho de armazenamento em nuvem dedicado à tabela.
schema_locationArmazena informações de rastreamento de esquema do Auto Loader.Use um caminho estável separado dos dados de origem e da tabela de destino.
Caminho de origemArquivos recebidos.Use um prefixo de aterrissagem específico da tabela quando possível.

Observabilidade de fluxo​

Use métricas de fluxo para verificar se a ingestão disponível agora processou arquivos e gravou linhas. ContractForge agrega resultados de microlotes e métricas de execução infantil na tabela de controle de fluxo.

SELECT
  stream_run_id,
  status,
  target_table,
  batches_processed,
  total_rows_read,
  total_rows_written,
  total_rows_quarantined,
  error_message
FROM main.ops.ctrl_ingestion_streams
ORDER BY started_at_utc DESC;

SELECT
  parent_run_id,
  run_id,
  status,
  rows_read,
  rows_written,
  rows_quarantined
FROM main.ops.ctrl_ingestion_runs
WHERE parent_run_id = '<stream_run_id>'
ORDER BY started_at_utc;

Qualidade e quarentena em microlotes​

As regras de qualidade se aplicam dentro de cada microlote. Se as linhas estiverem em quarentena, o resultado do lote filho registrará as métricas no nível da linha e o agregado do fluxo as somará.

quality_rules:
  not_null: [order_id]
  expressions:
    - name: positive_amount
      expression: payload.amount >= 0
      severity: quarantine
      message: Amount must be non-negative
on_quality_fail: quarantine

Problemas comuns​

SintomaCausa provávelAção
Arquivos não coletadosO checkpoint já os marcou como processados ​​ou includeExistingFiles é falso.Use um novo checkpoint para testes ou valide opções de cloudFiles.
Erros de esquemaDesvio de esquema inferido ou localização de esquema ausente.Declare dicas de esquema/esquema e use localização de esquema estável.
Stream diz zero linhas, mas o destino tem dadosBug de agregação de métricas ou incompatibilidade de execução secundária.Verifique ctrl_ingestion_runs.parent_run_id; as versões atuais agregam métricas secundárias.
Acesso negadoO runtime não pode acessar o caminho de origem/checkpoint/esquema.Corrija localização externa, concessões de volume ou credenciais de armazenamento de cluster.
Fluxo contínuo esperadoO contrato usa semântica disponível agora.Use um trabalho de streaming dedicado fora do ContractForge para fluxos sempre ativos.