Pipeline ETL automatizado que extrai dados estruturados de boletins diários em PDF usando parsing via LLM, orquestrado no Databricks com Arquitetura Medallion.
O JOX Pipeline ingere um boletim agropecuário diário (PDF), converte para Markdown, extrai notícias e cotações estruturadas via LLM (Claude Haiku 4.5) e persiste os resultados em Delta Tables no Databricks Unity Catalog — com uma camada Gold tipada pronta para consumo em BI.
Todo o pipeline roda em um único Databricks Notebook, agendado diariamente via Databricks Workflows.
PDF (web) → Markdown (Docling) → Extração LLM (Claude Haiku) → Delta Tables → Camada Gold (BI)
| Camada | Formato | Finalidade |
|---|---|---|
| Raw | PDF + Markdown em Unity Catalog Volumes | Dados-fonte imutáveis, trilha de auditoria completa |
| Silver | Delta Tables (noticias, cotacoes) |
Dados estruturados com histórico append e idempotência |
| Gold | Delta Tables (schema tipado) | Colunas tipadas (DATE, STRING), schema limpo para ferramentas de BI (Power BI, DAX) |
As iterações iniciais usavam 6 notebooks separados orquestrados via dbutils.notebook.run(). Isso causava:
restartPython()limpando o estado dos módulos entre notebooks- Mensagens de erro sendo engolidas nas fronteiras entre notebooks
- Gerenciamento complexo de dependências entre cells
A abordagem consolidada em notebook único (5 sprints sequenciais) resolveu tudo isso mantendo cada sprint claramente delimitado com headers Markdown.
A LLM retorna JSON livre. Os modelos Pydantic v2 (NewsItem, QuoteItem, ExtractionPayload) validam cada campo no momento da extração — capturando datas malformadas, campos ausentes e incompatibilidades de tipo antes dos dados chegarem à camada Delta. Um field_validator garante conformidade com datas ISO-8601.
Cotações frequentemente chegam como faixas (ex: "129,00/132,00"). Converter para numérico perderia essa informação. Analistas tratam o parsing em DAX/Power BI onde as regras de negócio para faixas já estão definidas.
O Docling (PDF → Markdown) inclui PyTorch, que pode causar OOM em clusters leves. O pipeline instala sob demanda apenas antes do Sprint 2, enquanto dependências mais leves (requests, pydantic, openai) são instaladas no início.
| Componente | Tecnologia |
|---|---|
| Orquestração | Databricks Notebooks + Workflows |
| PDF → Markdown | Docling (OCR opcional) |
| Extração LLM | Claude Haiku 4.5 via Databricks Serving Endpoint (API compatível OpenAI) |
| Validação de Schema | Pydantic v2 (structured output, auto-correção de JSON) |
| Armazenamento | Delta Tables no Unity Catalog / Volumes |
| Configuração | YAML com dataclasses imutáveis (frozen) |
| Logging | PipelineLogger customizado com métodos narrativos (section, step, detail) |
jox-llm-etl/
├── config/
│ └── config.yaml # Configuração do pipeline (paths, LLM, Delta, Gold)
├── notebooks/
│ └── pipeline_jox.py # Notebook principal do Databricks (5 sprints)
├── src/
│ ├── config/
│ │ └── settings.py # Loader de config via dataclasses imutáveis
│ ├── domain/
│ │ ├── schemas.py # Modelos Pydantic v2 (NewsItem, QuoteItem, ExtractionPayload)
│ │ ├── errors.py # Tipos de erro do domínio
│ │ └── results.py # Result pattern (ExtractionResult, WriteResult)
│ ├── downloader/
│ │ └── pdf_downloader.py # Downloader HTTP com lógica de retry
│ ├── infrastructure/
│ │ ├── llm_adapter.py # Adapter LLM compatível OpenAI + system prompt + reparo de JSON
│ │ ├── delta_writer.py # Writer para Delta Tables com check de idempotência
│ │ ├── docling_adapter.py # Wrapper de conversão Docling PDF→MD
│ │ └── markdown_reader.py # Leitor de arquivos Markdown
│ ├── logging/
│ │ ├── logger.py # PipelineLogger com métodos narrativos
│ │ └── formatters.py # Formatter plain (sem ANSI — compatível Databricks)
│ ├── pipeline/
│ │ ├── ingestion_pipeline.py # Sprint 1: Download PDF
│ │ ├── markdown_pipeline.py # Sprint 2: Conversão PDF → Markdown
│ │ ├── extraction_pipeline.py # Sprint 3: Extração via LLM
│ │ └── output_pipeline.py # Sprint 4: Persistência Delta
│ ├── services/
│ │ └── path_resolver.py # Resolução de paths baseada em data para Volumes
│ └── storage/
│ ├── file_storage.py # Abstração de I/O de arquivos
│ └── markdown_storage.py # Storage específico para Markdown
├── run_pipeline.py # Exemplo de configuração Databricks Jobs API
├── requirements.txt # Referência de dependências Python
└── .gitignore
O notebook executa 5 sprints sequenciais:
Baixa o PDF do boletim diário da URL fonte e salva em um Unity Catalog Volume. Inclui idempotência: pula se o arquivo do dia já existir (a menos que force_overwrite=true).
Instala o Docling sob demanda e converte o PDF para Markdown. Preserva estruturas de tabela para parsing downstream pela LLM.
Envia o Markdown para o Claude Haiku 4.5 via Databricks Serving Endpoint com um system prompt detalhado. A resposta é parseada como JSON, validada contra modelos Pydantic, e se a validação falhar, um loop de auto-correção reenvia o erro para a LLM corrigir (até 3 tentativas).
Entidades extraídas:
NewsItem— data, categoria (SUÍNOS, BOVINOS, FRANGO VIVO...), texto completoQuoteItem— data, grupo, produto, unidade, preço (string raw), status/tendência
Grava os DataFrames validados em Delta Tables com modo append e check de idempotência (evita linhas duplicadas para a mesma data).
Cria tabelas Gold tipadas via CREATE OR REPLACE TABLE ... AS SELECT com casts explícitos (DATE, STRING). Essas tabelas são o contrato para dashboards de BI.
Todas as configurações ficam em config/config.yaml. Seções principais:
# Fonte de dados
source:
url: "https://jox.com.br/boletim/diario.pdf"
# Paths no Unity Catalog
storage:
pdf_base_path: "/Volumes/<catalog>/jox/raw/pdf"
md_base_path: "/Volumes/<catalog>/jox/raw/md"
# Delta Tables (Silver)
delta_output:
catalog: "my_catalog"
schema: "jox"
# LLM (Databricks Serving Endpoint)
llm:
model: "databricks-claude-haiku-4-5"
databricks_host: "https://adb-<workspace-id>.<region>.azuredatabricks.net"
databricks_token_scope: "your-scope"
databricks_token_key: "your-token-key"
# Camada Gold
gold_output:
catalog: "my_catalog"
schema: "gold_jox"O token da LLM é obtido em runtime via dbutils.secrets.get(). Nenhuma credencial é armazenada no código ou nos arquivos de configuração.
- Workspace Databricks com Unity Catalog habilitado
- Serving endpoint com Claude Haiku 4.5 (ou qualquer modelo compatível OpenAI)
- Secret scope no Databricks com o token da API
- Cluster com Python 3.10+
- Importe o repositório no Workspace do Databricks via Git Folder
- Atualize
config/config.yamlcom seu catálogo, schema, host e secret scope - Crie um secret no Databricks:
databricks secrets put-secret --scope your-scope --key your-token-key
- Execute
notebooks/pipeline_joxa partir do Workspace - (Opcional) Agende como job diário via Databricks Workflows — veja
run_pipeline.pypara exemplo de configuração via Jobs API
| Coluna | Tipo | Descrição |
|---|---|---|
data |
STRING | Data do boletim (formato ISO) |
tipo |
STRING | Categoria da notícia |
noticia |
STRING | Texto completo da notícia |
| Coluna | Tipo | Descrição |
|---|---|---|
data |
STRING | Data do boletim (formato ISO) |
grupo |
STRING | Grupo do produto |
produto |
STRING | Nome do produto |
unid |
STRING | Unidade de medida |
valor_rs |
STRING | Preço em R$ (pode conter faixas) |
stat |
STRING | Indicador de tendência |
Este projeto é disponibilizado como referência de portfólio. Sinta-se à vontade para usar como inspiração para seus próprios pipelines.