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

Importante

Esse recurso está em Beta. Para usá-lo, um administrador de espaço de trabalho deve ativar o Tipo de Arquivo na página Visualizações. Consulte Gerenciar visualizações do Azure Databricks.

Aprenda a criar um pipeline do tipo medalhão com o pipeline do Lakeflow que processa documentos não estruturados de ponta a ponta. Este exemplo utiliza o samples.sec.contracts conjunto de dados de exemplo, uma coleção de acordos legais arquivados pela SEC armazenados como PDFs em um volume do Unity Catalog.

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

Para a referência tipográfica, veja FILE tipo.

Neste tutorial, você irá:

O resultado é um pipeline estilo medallion: bronze (referências gerenciadas FILE brutas), prata (documentos analisados e classificados) e ouro (campos extraídos por tipo de acordo). Consulte O que é a arquitetura do lakehouse de medalhão? para obter mais informações. A camada bronze é uma tabela de fluxo que ingere arquivos de forma incremental, e as camadas prata e ouro são vistas materializadas que só se recomputam quando suas entradas mudam.

Requirements

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

  • Esteja logado em um workspace do Azure Databricks com o Unity Catalog ativado.
  • Tenha o tipo FILE habilitado para o seu espaço de trabalho. Administradores do espaço de trabalho podem habilitá-lo pela página de Prévias . Consulte Gerenciar visualizações do Azure 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 esse volume como o FileSpace da tabela bronze, e o Unity Catalog copia os arquivos ingeridos nele como armazenamento gerenciado.
  • Use o canal Prévia.

O samples.sec.contracts conjunto de dados está disponível em todos os espaços de trabalho por padrão. Este tutorial armazena os PDFs ingeridos como referências FILE MANAGED: o Unity Catalog copia cada arquivo para o volume que você declara como o FileSpace da tabela e o gerencia com a tabela, então 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 etapas.

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

Use o Auto Loader para ler os PDFs dos contratos de forma incremental a partir do volume. A leitura de arquivos com format => 'file' captura uma referência e metadados para cada arquivo sem materializar os bytes. Declarar a coluna como FILE MANAGED copia cada arquivo para o volume da tabela FileSpace, o volume que você define com a databricks.filespace-preview propriedade tabela, então o Unity Catalog gerencia 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');

Python

from pyspark import pipelines as dp

@dp.table(
  name="raw_contracts",
  schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)
def raw_contracts():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .load("/Volumes/samples/sec/contracts/")
  )
  • Funciona para arquivos grandes: um PDF grande está na FileSpace da tabela, enquanto a linha da tabela armazena apenas uma referência leve FILE (uri, size, content_type, checksum). Compare isso com o tipo BINARY, que insere os bytes diretamente na linha.
  • Ciclo de vida gerenciado de arquivos: O Unity Catalog copia cada arquivo ingerido para o(a) FileSpace da tabela e o gerencia com a tabela: excluir linhas torna os arquivos referenciados elegíveis para coleta de lixo, assim a tabela e seus arquivos permanecem sincronizados. Para detalhes, veja FILE MANAGED e FILE EXTERNAL.
  • Processamento incremental: a tabela de streaming ingere progressivamente novos arquivos à medida que chegam à fonte, sem reprocessar os existentes. O samples.sec.contracts conjunto de dados neste exemplo é estático, mas com uma fonte ativa, novos arquivos são captados a cada atualização do pipeline. Para também propagar alterações e exclusões de fonte, ingira o feed de alterações com AUTO CDC. Veja Aplicar atualizações e exclusões com AUTO CDC.

Etapa 2. Prata: analisar e classificar documentos

Passe cada FILE para a ai_parse_document função para converter o PDF bruto em um VARIANT estruturado contendo elementos do documento, metadados de layout e texto. Como ai_parse_document aceita uma FILE coluna, 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;

Python

@dp.materialized_view(name="parsed_contracts")
def parsed_contracts():
  return (
    spark.read.table("raw_contracts")
      .selectExpr("path", "ai_parse_document(file) AS parsed")
  )

Observação

Definir a etapa de análise como uma vista materializada sobre a raw_contracts tabela de fluxo incrementa o cálculo. Cada atualização do pipeline é executada ai_parse_document apenas nos arquivos adicionados desde a última atualização, não na tabela inteira. Como ai_parse_document é a etapa mais cara, isso evita analisar novamente os documentos que você já processou. A atualização incremental das visualizações materializadas requer computação sem servidor; execute o pipeline em computação sem servidor. Consulte Pipelines Declarativos do Spark.

Em seguida, passe a saída processada para a ai_classifyfunção para atribuir a cada documento um dos cinco tipos de acordo. Documentos com erros de análise são filtrados antes da classificação. Este exemplo fixa ai_classify na versão 2.1, que retorna a classificação como um objeto por rótulo, portanto leia o rótulo na chave 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);

Python

@dp.materialized_view(name="classified_contracts")
def classified_contracts():
  return (
    spark.read.table("parsed_contracts")
      .filter("is_variant_null(parsed:error_status)")
      .selectExpr(
        "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""")
  )

Dica

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

Etapa 3. Ouro: extrair campos por tipo de acordo

Cada tipo de acordo possui seu próprio conjunto de campos relevantes. Filtre os documentos classificados para um único tipo, passe o conteúdo analisado para a ai_extract função com o esquema dos campos que você deseja e, em seguida, transforme a resposta em colunas tipadas. Este exemplo fixa ai_extract na versão 2.1, em que cada campo extraído é um objeto; portanto, leia a chave value dele.

O exemplo a seguir cria a tabela Gold para acordos 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;

Python

@dp.materialized_view(name="consulting_agreements")
def consulting_agreements():
  return (
    spark.read.table("classified_contracts")
      .filter("contract_type = 'consulting_agreement'")
      .selectExpr(
        "path",
        """ai_extract(
             parsed,
             '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
             map('version', '2.1')
           ) AS fields""")
      .selectExpr(
        "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")
  )

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

Exemplos de notebooks

Os cadernos a seguir contêm o pipeline completo deste tutorial. Esses notebooks são código-fonte de pipeline, não notebooks executáveis. Importe o notebook no seu idioma e especifique o caminho dele no campo Código-fonte quando configurar o pipeline. Consulte Configurar pipelines.

SQL

Notebook SQL do pipeline de processamento de arquivos

Obter laptop

Python

notebook Python para pipeline de processamento de arquivos

Obter laptop

Explore por conta própria

O pipeline classifica documentos em cinco tipos de contrato, mas extrai campos apenas para consulting_agreement. Para estendê-lo, repita a etapa gold para cada tipo restante, alterando o filtro contract_type e o esquema ai_extract para que correspondam aos campos relevantes para esse tipo. Por exemplo:

  • affiliate_agreement: party_1_name, party_2_name, , commission_ratepayment_frequency
  • marketing_agreement: party_1_name, party_2_name, , effective_dateterritory
  • hosting_agreement: provider_name, customer_name, , effective_dateterm_length
  • escrow_agreement: owner_name, licensee_name, , escrow_agent_namesoftware_name

Recursos adicionais

  • Última atualização em