ETL incremental em Python com Pandas, DuckDB e Parquet (checkpoints)

Quando e por que usar

ETL incremental não é um luxo — é uma necessidade quando os volumes crescem e você precisa reduzir I/O, tempo de processamento e custo. Em vez de reprocessar toda a base, você foca apenas nas linhas novas ou modificadas desde o último processamento. Aqui vamos trabalhar com pandas para transformação, pyarrow/parquet para armazenamento columnar e duckdb para consultas rápidas e compactação/compaction.

Problema prático

Suponha uma fonte que emite linhas com um campo incremental (id) ou um carimbo de tempo (updated_at). Reprocessar tudo diariamente pode ser inviável. O padrão incremental usa um checkpoint (arquivo simples) com a última chave processada e um pipeline que filtra as novas linhas e as incorpora no destino em parquet.

Requisitos mínimos

Para seguir os exemplos você precisa de Python 3.8+, pandas, duckdb, pyarrow e opcionalmente sqlalchemy para fontes SQL. Instale com: pip install pandas duckdb pyarrow sqlalchemy. Use ambientes virtuais. Vamos assumir permissões de I/O em disco e que parquet é um formato aceito no pipeline.

Comparação com alternativas

Existem várias abordagens para ETL incremental: CDC (Change Data Capture) via Kafka, frameworks como Airflow/Prefect, ferramentas proprietárias (Fivetran), e scripts customizados. A solução que mostramos é focada em simplicidade operacional, auditabilidade e custo baixo, adequada para times pequenos.

Prós e contras vs CDC e messagens

CDC é robusto para latência baixa e integridade, mas adiciona infra (Debezium, Kafka). Nosso método é batch-incremental: mais simples, menos infra e latência na ordem de minutos/hours, boa para analytics e relatórios.

Por que DuckDB + Parquet

Parquet é eficiente para análises columnar e compressão. DuckDB consulta parquet diretamente e permite operações de compaction, joins e dedup sem mover dados para um DW. Isso reduz custo e aumenta velocidade em workloads analíticos pequenos/median.

Implementação progressiva

Vamos implementar um pipeline incremental passo a passo: (1) checkpoint, (2) extrair apenas novos registros, (3) transformar em pandas, (4) escrever append em parquet com partição, (5) usar DuckDB para compactar e garantir idempotência.

Checkpoint simples (leitura e escrita)

Checkpoint é um arquivo JSON com a chave last_id ou last_ts. Mantê-lo simples facilita debugging e rollbacks. Exemplo abaixo: utilitário que lê e atualiza o checkpoint.

import json
from pathlib import Path

# O que faz: lê o checkpoint atual e atualiza com novo valor.
# Uso: ler antes da extração, atualizar depois do append bem-sucedido.

CHK_PATH = Path('checkpoints/last_checkpoint.json')

def read_checkpoint():
    if not CHK_PATH.exists():
        return {'last_id': 0}
    return json.loads(CHK_PATH.read_text())

def write_checkpoint(state: dict):
    CHK_PATH.parent.mkdir(parents=True, exist_ok=True)
    CHK_PATH.write_text(json.dumps(state))

# O que acontece: retorna dict com 'last_id' e escreve o arquivo.

Explicação: usar JSON facilita inspeção manual. Em produção você pode usar S3/DB como checkpoint storage, mas a lógica permanece a mesma: atomicidade na escrita é crítica (escrever em arquivo temporário e renomear é uma boa prática).

Extração incremental de uma fonte SQL

Exemplo: extrair novos registros de uma tabela events usando last_id. Implementamos uma consulta parametrizada que só busca linhas com id > last_id ordenadas. Aqui usamos sqlalchemy como exemplo, mas o mesmo vale para leitura via CSV ou APIs.

import pandas as pd
from sqlalchemy import create_engine, text

# O que faz: conecta ao banco, busca novos registros desde last_id.

def extract_new_rows(conn_str: str, last_id: int) -> pd.DataFrame:
    engine = create_engine(conn_str)
    query = text('SELECT id, user_id, event_type, payload, created_at FROM events WHERE id > :last_id ORDER BY id')
    with engine.connect() as conn:
        df = pd.read_sql(query, conn, params={'last_id': last_id})
    return df

# O que acontece: retorna um DataFrame com novas linhas (pode ser vazio).

Explicação: ordenar por id garante que o último id no DataFrame é o novo checkpoint. Em ambientes com múltiplas fontes, normalize e adicione coluna source_id para evitar colisões.

Transformação e validações com pandas

Transforme usando pandas: normalizar timestamps, validar esquemas e remover duplicatas. Mantenha validações rápidas e falhe cedo. Exemplo de pequenas transformações e validação do schema.

from datetime import datetime

# O que faz: normaliza campos e valida tipos mínimos

def transform(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df
    df = df.copy()
    df['created_at'] = pd.to_datetime(df['created_at'], utc=True)
    df['event_type'] = df['event_type'].astype('category')
    # valida: id não nulo e único no batch
    assert df['id'].isnull().sum() == 0
    assert df['id'].duplicated().sum() == 0
    return df

# O que acontece: retorna DataFrame pronto para escrita em parquet.

Explicação: cast para category reduz memória antes de gravar parquet. Use asserts para falhar rapidamente durante desenvolvimento; em produção, converta a excessão para alertas e rollback do checkpoint.

Append em Parquet com pyarrow e particionamento

Append em parquet precisa de atenção: pandas.DataFrame.to_parquet não tem append padrão para datasets particionados. Usamos pyarrow.parquet.write_to_dataset para escrever em modo dataset com partition_cols. Isso facilita consultas por partição e compaction posterior.

import pyarrow as pa
import pyarrow.parquet as pq

# O que faz: escreve DataFrame para um dataset parquet particionado por ano/mês

def append_parquet(df: pd.DataFrame, base_path: str):
    if df.empty:
        return
    table = pa.Table.from_pandas(df)
    pq.write_to_dataset(table, root_path=base_path, partition_cols=['year', 'month'])

# Exemplo de uso: antes de chamar, crie colunas year/month:
# df['year'] = df['created_at'].dt.year
# df['month'] = df['created_at'].dt.month

# O que acontece: arquivos parquet são escritos em base_path/year=YYYY/month=MM/...

Explicação: este método cria novos ficheiros por chamada. Depois de múltiplos pequenos arquivos, faça compaction com DuckDB para otimizar leitura.

Integracao com ecossistema real

A integração prática inclui: orquestração (cron, Airflow, Prefect), armazenamento (S3/MinIO), e monitoramento (logs, alertas). Vamos focar em compactação, deduplicação e criação de views com DuckDB, que permite operar diretamente sobre arquivos parquet.

Compactação e deduplicação com DuckDB

Depois de muitos pequenos arquivos, use DuckDB para ler o dataset, deduplicar por id e reescrever em um snapshot otimizado. O exemplo abaixo cria um snapshot parquet único (ou particionado) com commit atômico.

import duckdb
from pathlib import Path

# O que faz: cria um snapshot compactado (overwrite) a partir dos parquet shards

def compact_parquet(source_pattern: str, target_path: str, temp_path: str):
    Path(temp_path).mkdir(parents=True, exist_ok=True)
    con = duckdb.connect()
    # lê todos os parquet e deduplica por id mantendo o maior created_at
    sql = f"""
    CREATE TABLE tmp AS
    SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY created_at DESC) as rn
    FROM read_parquet('{source_pattern}');
    """
    con.execute(sql)
    con.execute('CREATE TABLE dedup AS SELECT * EXCLUDE(rn) FROM tmp WHERE rn = 1')
    # escreve para parquet otimizado
    con.execute(f"COPY dedup TO '{temp_path}/snapshot.parquet' (FORMAT PARQUET);")
    # mover atomicamente
    Path(temp_path, 'snapshot.parquet').replace(Path(target_path))
    con.close()

# O que acontece: target_path recebe um parquet otimizado com registros deduplicados.

Explicação: usar read_parquet do DuckDB simplifica consultas SQL em arquivos. A técnica ROW_NUMBER permite deduplicar mantendo a versão mais recente.

Consumidores: FastAPI e consultoria de leitura

Se você expõe dados para APIs, use DuckDB para consultas ad-hoc e sirva resultados via FastAPI. DuckDB é embutível e rápido para conjuntos analíticos. Exemplo mínimo: endpoint que roda uma query parametrizada em parquet e retorna JSON.

Arquitetura recomendada: dados brutos em S3 como parquet; processos ETL incrementais atualizam os shards; compaction programado cria snapshots otimizados; DuckDB/Presto/Trino consomem os snapshots para APIs/BI.

Checklist de qualidade para producao

Abaixo um checklist prático para mover esse pipeline para produção. Cada item é testável e auditável.

Operacional

• Logs estruturados (JSON) com request_id e checkpoints.
• Retries idempotentes: garantir que re-execução não duplica dados; checkpoint só gravado após sucesso.
• Alertas para batches vazios anormais ou crescimento de latência.

Dados e integridade

• Validações de schema na entrada (tipos, nullability).
• Checksums ou counts antes/depois do append para detectar perdas.
• Testes de deduplicação e integridade após compaction.

Exemplo: rotina completa que junta tudo

A seguir um script minimal que combina checkpoint, extração, transformação, append e checkpoint update. Use como base; adapte conexões e paths para produção.

def run_incremental(conn_str, base_parquet_path):
    state = read_checkpoint()
    last_id = state.get('last_id', 0)

    df = extract_new_rows(conn_str, last_id)
    df = transform(df)
    if df.empty:
        print('No new rows')
        return

    # preparar colunas para particionamento
    df['year'] = df['created_at'].dt.year
    df['month'] = df['created_at'].dt.month

    append_parquet(df, base_parquet_path)

    # atualizar checkpoint com o maior id processado
    new_last = int(df['id'].max())
    write_checkpoint({'last_id': new_last})
    print('Batch processed, new last_id=', new_last)

# O que acontece: processa apenas registros > last_id e atualiza checkpoint no final.

Explicação: sempre atualizar checkpoint só após append bem-sucedido. Em ambientes distribuídos, um lock simples (file lock) evita concorrência.

Erros comuns, armadilhas e dicas práticas

Pipeline incremental costuma falhar por erros previsíveis. Conhecê-los evita retrabalho e perda de dados.

Checkpoint escrito antes do append

Armzalha: gravar o checkpoint antes de garantir que os dados foram persistidos. Resultado: perda de dados. Sempre escrever checkpoint apenas após confirmação explícita da escrita e validação (checksum ou contagem).

Pequenos arquivos parquet (file small write problem)

Escrever muitos arquivos pequenos degrada performance. Solução: agrupar writes em batches maiores, agendar compaction periódica com DuckDB e usar particionamento eficiente (ano/mês ou dia dependendo do volume).

Dica: escreva em um diretório temp e só mova/commit o arquivo para o diretório final quando estiver pronto. Isso evita readers verem estados parciais.

Schema drift

Alterações de schema (nova coluna, rename) quebram leitura. Mitigue com validações: comparar colunas esperadas e, se houver variação, lançar alerta e tomar roteamento manual ou transformação compatível (preencher colunas faltantes com nulls).

Conclusão prática

O padrão incremental com pandas + parquet + duckdb oferece uma solução de baixo custo e alto rendimento para pipelines analíticos. Checkpoints simples garantem idempotência; pyarrow escreve parquet eficientes; DuckDB permite compaction e queries SQL sobre os dados no próprio armazenamento. Combine esses blocos e você terá um pipeline auditável, reprodutível e fácil de operar.

Resumo das ações concretas: (1) mantenha checkpoints imutáveis até confirmação; (2) extraia apenas o delta usando last_id/last_ts; (3) escreva parquet em batches com partition_cols; (4) compacte e dedup com DuckDB; (5) monitore e alerte sobre batches vazios e tempos de execução.

Comece com scripts locais e evolua para orquestração se a complexidade aumentar. A maior vantagem dessa pilha é a simplicidade e observabilidade — você consegue inspecionar todos os artefatos (arquivos parquet, checkpoints) diretamente no disco ou S3.

Próximos passos recomendados: implementar testes end-to-end (gerar dados, rodar pipeline, validar counts), adicionar métricas (batch_duration, rows_processed) e proteger checkpoint storage com versionamento (S3 object versioning ou DB transactional table).