ZrobDev/pipelines-ETL

Pipelines ETL

Ingestão em lote no PostgreSQL (armazenamento padrão) com Python (uv, ruff, pytest). Transformações locais: polars 2.x ou DuckDB 2.0.

Procedimento

1. Defina o grão: uma linha significa o quê, identificada por qual chave natural. Sem chave, não há carga idempotente; defina uma (ou um hash de conteúdo) primeiro.
2. Escolha o padrão de carga (abaixo). Escolha a estratégia incremental. Escolha o orquestrador por último.
3. Escreva a execução de forma que reexecutar qualquer janela produza o mesmo estado final. Prove isso: execute duas vezes em um teste e afirme que as contagens de linhas e um checksum permanecem inalterados.
4. Adicione as verificações de observabilidade (abaixo) antes da primeira execução em produção.
5. Revise usando a checklist no final.

Padrão de carga idempotente

• Linhas têm uma chave natural estável e podem mudar: INSERT ... ON CONFLICT (key) DO UPDATE (Postgres) ou MERGE. Adicione WHERE target.hash IS DISTINCT FROM excluded.hash para que linhas inalteradas não causem rotação.
• Dados chegam como uma partição inteira (dia, id do lote) e são regravados como uma unidade: exclua-insira essa partição em UMA transação. Uma exclusão confirmada separadamente da inserção deixa um buraco em caso de falha.
• Tabela inteira é reconstruída: carregue em uma tabela temporária, valide, depois troque (renomeie em uma transação). Leitores nunca veem uma tabela meio carregada.
• Fatos somente anexados sem chave: adicione um batch_id / hash do arquivo de origem determinístico e faça com que a carga recuse ou substitua um id de lote já carregado.
• Nunca faça INSERT cegamente e depois deduza. Duplicatas chegam a um painel antes da deduplicação.
• Carregue via COPY para uma tabela temporária/estágio e depois faça um upsert baseado em conjunto. Upserts linha por linha a partir do Python são 10 a 100 vezes mais lentos.

Cargas incrementais

• Watermark em uma coluna que a origem garante ser monotônica e indexada (updated_at, id de sequência). Armazene-a em uma tabela de estado, escrita na mesma transação da carga, nunca em um arquivo separado.
• Leia watermark - lookback (ex.: 1 a 24 h), não > watermark. As confirmações da origem não são ordenadas por timestamp; o lookback mais o upsert absorve linhas tardias. Ajuste o tamanho do lookback com base no atraso medido, não em um palpite.
• Coluna de watermark não confiável (não atualizada em update, timezone misto, soft deletes ausentes): faça uma reconciliação completa periódica (diff de conjunto de chaves ou contagem + checksum por partição) em vez de confiar nela.
• Exclusões físicas na origem são invisíveis para watermarks. Decida explicitamente: reconcilie por diff de chave, ou aceite e documente.
• CDC (replicação lógica do Postgres, leitores de log estilo Debezium): use apenas quando precisar de exclusões, latência inferior a um minuto, ou a origem não puder ser consultada por timestamp. Isso adiciona slots, risco de retenção (um slot ocioso enche o disco da origem) e problemas de ordenação/tratamento de snapshot. Caso contrário, permaneça com watermarks.

Backfills

• Todo job aceita --start e --end (semi-aberto [start, end), UTC) e deriva tudo deles. Nenhum now() dentro da transformação.
• Um backfill é a execução normal do job sobre janelas antigas. Se precisar de código diferente, o job normal não é idempotente; corrija isso.
• Divida em pedaços por partição, execute os pedaços independentemente, registre o status por pedaço, para que uma falha no pedaço 40 retome do 40.
• Faça o backfill no mesmo destino usando o mesmo upsert/substituição de partição. Verifique se não há contagem duplicada: count(*) e count(distinct key) por janela antes e depois.
• Faça throttling contra a API/BD de origem; um backfill é um pico de carga autoinfligido.

Retries, timeouts, "exatamente uma vez"

• Entrega exatamente uma vez não existe através de uma rede. Construa entrega pelo menos uma vez mais escritas idempotentes; o resultado é efetivamente uma vez.
• Retry apenas em etapas idempotentes, com backoff exponencial e jitter, um limite de tentativas e um timeout rígido em cada chamada de rede (conexão e leitura configurados separadamente). Retry em 429/5xx/timeouts, não em 4xx.
• Respeite Retry-After. Retries sem um limite global de taxa amplificam uma interrupção.
• Uma etapa que envia e-mail, cobrança ou posta em um webhook não é idempotente: proteja com uma chave de idempotência armazenada antes da chamada.
• Em caso de falha final, saia com código diferente de zero e alerte; nunca engula a exceção para manter um cronograma verde.

Desvio de esquema

• A camada de chegada é permissiva (armazene o payload bruto como JSON/texto mais metadados de carga). A camada tipada é estrita e falha ruidosamente em violações de tipo/campo obrigatório.
• Novas colunas: aditivas e permitidas com um alerta. Colunas removidas, renomeadas ou alteradas de tipo: falhe a execução e exiba a diferença; não force para NULL.
• Valide com um esquema explícito (schema do polars, pydantic, ou DuckDB DESCRIBE) na fronteira, uma vez, não espalhado pela transformação.

Raspagem para warehouse

• Dois estágios. Estágio 1 busca e armazena a resposta bruta inalterada e imutável (caminho do objeto chaveado por origem, timestamp da busca, hash de conteúdo). Estágio 2 analisa o bruto em tabelas.
• Bugs no parser são então corrigidos reanalisando o bruto armazenado, sem nova busca, sem risco de bloqueio e sem perda de histórico.
• Registre URL, timestamp da busca, status HTTP, versão do parser em cada linha analisada. A análise é uma função pura dos bytes brutos mais a versão do parser.
• Falhas na busca e páginas vazias/bloqueadas (captcha, 200 com muro de login) são detectadas por verificações de conteúdo, não apenas pelo código de status.

Observabilidade (mínimo)

• Registre por etapa: linhas lidas, linhas escritas, linhas rejeitadas, duração, início/fim da janela, id da execução. Linhas estruturadas (JSON).
• Afirme invariantes: linhas de saída vs entrada dentro de uma tolerância; zero linhas em uma janela que normalmente não é vazia FALHA a execução. Um sucesso silencioso de zero linhas é a falha não detectada mais comum.
• Heartbeat: envie um ping para uma verificação de saúde externa no sucesso e no início; alerte em um ping perdido (captura cron não disparando, host inativo). Falha também envia ping para o endpoint de falha.
• Alerte sobre frescor (timestamp máximo carregado mais antigo que o limite), não apenas em erros de job.
• Mantenha uma tabela de log de execuções (id da execução, janela, status, contagens); ela também é seu livro-razão de backfill.

Escolhendo o orquestrador

• Cron (ou timer systemd) + um script: um a poucos jobs independentes, sem dependências entre jobs, uma equipe que pode ler logs. Padrão. Precisa do heartbeat acima e de um lock (flock) contra execuções sobrepostas.
• n8n: baixo volume, cola entre APIs SaaS, acionado por webhook, não-desenvolvedores editam fluxos. Inadequado para grandes lotes, transformações pesadas ou revisão de código; mantenha a lógica pesada em um script Python que ele chama.
• Airflow 3.x: muitos DAGs, dependências entre jobs, backfills de primeira classe, retries e SLAs, UI de auditoria. Custo operacional real (scheduler, metadata DB, workers); não vale a pena para um punhado de jobs. A migração do Airflow 2 para 3 tem mudanças disruptivas; verifique a versão antes de copiar um trecho. Para dbt no Airflow use Astronomer Cosmos.
• dlt: ingestão de API ou banco de dados onde você quer inferência de esquema, estado incremental e disposições merge/replace tratadas para você. Use-o em vez de escrever manualmente código de extração paginada + watermark + upsert; ainda aplique as regras acima à sua configuração.
• Contêineres: segredos como segredos do Compose lidos de /run/secrets/, nunca env vars ou env_file; publique portas vinculadas a 127.0.0.1 ou mantenha-as em uma rede interna.

Skills de fornecedores

Quando a tarefa precisar de especificidades do dbt ou Airflow (modelos, testes, criação de DAG, operadores, configuração do Cosmos), não improvise: use os skills oficiais dos fornecedores puxados através do skill de-vendors deste repositório, e recorra à versão em references/landscape-2026-11.md apenas para verificar o que é atual.

Checklist de revisão

• Chave e grão definidos; reexecução da mesma janela produz estado idêntico (testado).
• Exclusão e inserção em uma transação; watermark confirmado com a carga.
• Lookback presente; dados tardios e exclusões têm uma resposta explícita.
• Timeouts, retries limitados com backoff, etapas não idempotentes protegidas.
• Chegada bruta imutável; análise separável e reexecutável.
• Contagens de linhas por etapa registradas; alertas de zero linhas e frescor existem; heartbeat configurado.
• Desvio de esquema falha ruidosamente na camada tipada.
• Escolha do orquestrador justificada pela quantidade de jobs e dependências, não pelo hábito.

Esqueletos SQL e Python concretos: references/patterns.md. Lições de uso real: references/gotchas.md.