Pular para o conteúdo principal
Última atualização em

Tutorial: criar um pipeline de processamento de arquivos com o tipo FILE

info

Beta

Este recurso está em Beta. Os administradores do Workspace podem controlar o acesso a este recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Saiba como criar um pipeline medalhão com o Lakeflow pipeline que processa documentos não estruturados de ponta a ponta. Este exemplo usa o dataset de exemplo samples.sec.contracts, uma coleção de contratos legais arquivados na SEC armazenados como PDFs em um volume do Unity Catalog.

O pipeline ingere os PDFs como referências FILE gerenciadas com o Auto Loader, analisa cada documento com funções de IA, classifica-o em um tipo de contrato e extrai campos estruturados para cada tipo.

Para a referência de tipo, consulte tipoFILE.

Neste tutorial, você:

O resultado é um pipeline no estilo medalhão: bronze (referências gerenciadas brutas FILE), silver (documentos analisados e classificados) e ouro (campos extraídos por tipo de contrato). Consulte Qual é a arquitetura medalhão do lakehouse? para obter mais informações. A camada bronze é uma tabela de transmissão que ingere arquivos incrementalmente, e as camadas silver e ouro são views materializadas que recomputam apenas quando suas entradas são alteradas.

Requisitos​

Para completar este tutorial, você deve atender aos seguintes requisitos:

  • Esteja logado em um workspace do Databricks com o Unity Catalog habilitado.
  • Tenha o tipo FILE ativado para o seu workspace. Os administradores do workspace podem ativá-lo na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.
  • Ter permissões para criar tabelas em um esquema e para criar um pipeline.
  • Tenha um volume do Unity Catalog no qual você possa gravar. Você declara este volume como o FileSpace da tabela bronze, e o Unity Catalog copia os arquivos ingeridos para ele como armazenamento gerenciado.
  • Use o canal de pré-visualização.

O dataset samples.sec.contracts está disponível em todos os workspaces por default. Este tutorial armazena os PDFs ingeridos como referências FILE MANAGED: o Unity Catalog copia cada arquivo para o volume que você declara como FileSpace da tabela e o gerencia com a tabela, portanto, excluir linhas torna os arquivos referenciados elegíveis para coleta de lixo e a tabela e seus arquivos permanecem sincronizados. Para adaptar o pipeline aos seus próprios PDFs, aponte o caminho de origem para um volume que contenha seus arquivos. Para outras opções de ingestão, consulte Ingerir arquivos como o tipo FILE.

Criar o pipeline de processamento de arquivos​

O pipeline processa documentos em três estágios.

Etapa 1. Bronze: ingerir PDFs brutos como referências de ARQUIVO gerenciadas​

Use o Auto Loader para ler incrementalmente os PDFs de contrato do volume. A leitura de arquivos com format => 'file' captura uma referência e metadados para cada arquivo sem materializar seus bytes. Declarar a coluna como FILE MANAGED copia cada arquivo para o FileSpace da tabela, o volume que você define com a propriedade de tabela databricks.filespace-preview, para que o Unity Catalog gerencie os arquivos com a tabela.

SQL
CREATE OR REFRESH STREAMING TABLE raw_contracts (
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
)
TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
AS SELECT *
  FROM STREAM read_files(
    '/Volumes/samples/sec/contracts/',
    format => 'file');
  • Funciona para arquivos grandes : um PDF grande reside no FileSpace da tabela, enquanto a linha da tabela armazena apenas uma referência FILE leve (uri, size, content_type, checksum). Compare isso com o tipo BINARY, que coloca os bytes em linha na linha.
  • Ciclo de vida de arquivo gerenciado : o Unity Catalog copia cada arquivo ingerido para o FileSpace da tabela e o gerencia com a tabela: a exclusão de linhas torna os arquivos referenciados elegíveis para coleta de lixo, para que a tabela e seus arquivos permaneçam sincronizados. Para obter detalhes, consulte FILE MANAGED e FILE EXTERNAL.
  • Processamento incremental : a tabela de transmissão ingere incrementalmente novos arquivos à medida que chegam à origem, sem reprocessar os existentes. O dataset samples.sec.contracts neste exemplo é estático, mas com uma origem ativa, novos arquivos são capturados a cada atualização do pipeline. Para também propagar alterações e exclusões de origem, ingira o feed de alterações com AUTO CDC. Consulte Aplicar atualizações e exclusões com AUTO CDC.

O passo 2. Silver: analisar e classificar documentos​

Passe cada FILE para a funçãoai_parse_document para converter o PDF bruto em um VARIANT estruturado contendo elementos do documento, metadados de disposição e texto. Como ai_parse_document aceita uma coluna FILE, ele lê o documento diretamente do armazenamento e nunca carrega os bytes na memória do cluster.

SQL
CREATE OR REFRESH MATERIALIZED VIEW parsed_contracts AS
  SELECT
    path,
    ai_parse_document(file) AS parsed
  FROM raw_contracts;
nota

Definir o passo de análise como uma view materializada sobre a tabela de transmissão raw_contracts torna a computação incremental. Cada atualização de pipeline realiza a execução de ai_parse_document apenas nos arquivos adicionados desde a última atualização, não na tabela inteira. Como ai_parse_document é o passo mais caro, isso evita reanalisar documentos que você já processou. O refresh incremental de views materializadas requer compute Serverless; faça a execução do pipeline em Serverless. See Spark Declarative Pipelines.

Em seguida, passe a saída analisada para a funçãoai_classify para atribuir a cada documento um dos cinco tipos de contrato. Documentos com erros de análise são filtrados antes da classificação. Este exemplo faz o pin de ai_classify para a versão 2.1, que retorna a classificação como um objeto por rótulo, portanto, leia o rótulo da key value.

SQL
CREATE OR REFRESH MATERIALIZED VIEW classified_contracts AS
  SELECT
    path,
    parsed,
    ai_classify(
      parsed,
      '["affiliate_agreement", "marketing_agreement", "consulting_agreement", "hosting_agreement", "escrow_agreement"]',
      map('version', '2.1')
    ):response[0].value::STRING AS contract_type
  FROM parsed_contracts
  WHERE is_variant_null(parsed:error_status);
dica

Para melhorar a precisão da classificação, adicione descrições de rótulos e uma opção instructions a ai_classify. Consulte a função ai_classify.

O passo 3. Ouro: extrair campos por tipo de acordo​

Cada tipo de contrato tem seu próprio conjunto de campos relevantes. Filtre os documentos classificados para um tipo, passe o conteúdo analisado para a funçãoai_extract com um esquema dos campos desejados e, em seguida, achate a resposta em colunas tipadas. Este exemplo fixa ai_extract na versão 2.1, na qual cada campo extraído é um objeto, portanto, leia sua key value.

O exemplo a seguir cria a tabela ouro para contratos de consultoria:

SQL
CREATE OR REFRESH MATERIALIZED VIEW consulting_agreements AS
  WITH extracted AS (
    SELECT
      path,
      ai_extract(
        parsed,
        '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
        map('version', '2.1')
      ) AS fields
    FROM classified_contracts
    WHERE contract_type = 'consulting_agreement'
  )
  SELECT
    path,
    fields:response.company_name.value::STRING AS company_name,
    fields:response.consultant_name.value::STRING AS consultant_name,
    fields:response.compensation_amount.value::STRING AS compensation_amount,
    fields:response.effective_date.value::STRING AS effective_date
  FROM extracted;

Com essas instruções, você tem um pipeline totalmente incremental: à medida que novos PDFs de contrato chegam ao volume, o Auto Loader os ingere como referências FILE gerenciadas, ai_parse_document e ai_classify roteiam cada documento, e a view materializada ouro consulting_agreements exibe os campos extraídos.

Notebooks de exemplo​

Os notebooks a seguir contêm o pipeline completo deste tutorial. Estes notebooks são código-fonte de pipeline, não notebooks executáveis. Importe o notebook para o seu idioma e, em seguida, especifique seu caminho no campo Source code ao configurar o pipeline. Consulte Configurar pipelines.

Notebook SQL de pipeline de processamento de arquivos

Abrir notebook em uma nova abaOpen in Databricks

Explore por conta própria​

O pipeline classifica documentos em cinco tipos de acordo, mas extrai campos apenas para consulting_agreement. Para estendê-lo, repita o passo ouro para cada tipo restante, alterando o filtro contract_type e o esquema ai_extract para corresponder aos campos relevantes para esse tipo. Por exemplo:

  • affiliate_agreement: party_1_name, party_2_name, commission_rate, payment_frequency
  • marketing_agreement: party_1_name, party_2_name, effective_date, territory
  • hosting_agreement: provider_name, customer_name, effective_date, term_length
  • escrow_agreement: owner_name, licensee_name, escrow_agent_name, software_name

Recursos adicionais​

Nesta página