Noções Básicas de Engenharia de Dados

9 exemplos para você começar com Engenharia de Dados - 7 básicos e 2 intermediários.

Pré-requisitos

• Python 3.14.0 com pandas 2.2+, pyarrow e familiaridade com orquestração.
• Configuração rápida: uv pip install pandas pyarrow prefect

Exemplos Básicos

1. Extrair para Parquet

Armazene dados brutos em armazenamento colunar antes das transformações.

import pandas as pd
 
df = pd.read_csv("orders.csv", parse_dates=["ordered_at"], dtype={"region": "category"})
df.to_parquet("raw/orders.parquet", index=False, compression="zstd")

• Parquet incorpora o esquema e comprime melhor que CSV.
• Caminhos de partição como dt=2025-01-15/ aceleram leituras seletivas.
• Nunca sobrescreva o raw - anexe partições por data de execução.

Relacionado: Formatos de Arquivo - Parquet e Arrow

2. Transformar com pandas

Limpe e modele dados em uma etapa scriptável.

import pandas as pd
 
df = pd.read_parquet("raw/orders.parquet")
clean = (
    df.dropna(subset=["order_id"])
    .assign(revenue=pd.to_numeric(df["revenue"], errors="coerce"))
    .query("revenue >= 0")
)
clean.to_parquet("staging/orders_clean.parquet", index=False)

• Cada estágio grava um artefato que você pode auditar.
• Registre contagens de linhas de entrada/saída em cada etapa.
• Mantenha regras de negócio em Python versionado.

Relacionado: Limpeza e Transformação de Dados

3. Carga Idempotente

Upsert ou sobrescrita de partição para que novas tentativas não dupliquem.

import pandas as pd
from pathlib import Path
 
run_dt = "2025-01-15"
out = Path(f"mart/orders/dt={run_dt}/data.parquet")
out.parent.mkdir(parents=True, exist_ok=True)
df.to_parquet(out, index=False)  # sobrescreve a partição para esta dt

• A chave de partição escopa a idempotência para um dia.
• Reexecutar o mesmo run_dt substitui, não anexa duplicados.
• Documente a política de dados que chegam atrasados separadamente.

Relacionado: Confiabilidade do Fluxo de Trabalho - decisões de idempotência

4. Agendar com cron

Jobs em batch rodam em um relógio quando a tolerância à latência é de horas.

# crontab: 0 6 * * * /path/.venv/bin/python /path/jobs/daily_orders.py
import logging
logging.basicConfig(level=logging.INFO)
logging.info("daily_orders started")

• Cron é suficiente para pipelines MVP de servidor único.
• Redirecione logs para arquivos ou logging centralizado.
• Saia com código diferente de zero em caso de falha para que o monitoramento alerte.

Relacionado: Airflow - agendamento de DAGs

5. Limites de Tarefa

Uma função por etapa com entradas/saídas tipadas.

from pathlib import Path
import pandas as pd
 
def extract(path: Path) -> pd.DataFrame:
    return pd.read_parquet(path)
 
def transform(df: pd.DataFrame) -> pd.DataFrame:
    return df.groupby("region", observed=True)["revenue"].sum().reset_index()
 
def load(df: pd.DataFrame, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(path, index=False)

• Funções puras simplificam testes unitários.
• Passe caminhos ou configurações - evite globais ocultos.
• O orquestrador conecta extract -> transform -> load.

Relacionado: Prefect & Dagster

6. Validar Esquema

Rejeite dados ruins antes que cheguem aos marts.

import pandera.pandas as pa
 
schema = pa.DataFrameSchema({
    "order_id": pa.Column(int, unique=True),
    "revenue": pa.Column(float, pa.Check.ge(0)),
})
schema.validate(df)

• Falhas de validação devem bloquear tarefas downstream.
• Armazene Parquet de quarentena para linhas rejeitadas.
• Versionar esquemas junto com o código do pipeline.

Relacionado: Validação e Qualidade de Dados

7. Registrar Contagens de Linhas

Visibilidade operacional começa com métricas simples.

import logging
logger = logging.getLogger(__name__)
 
def transform(df):
    logger.info("input_rows=%s", len(df))
    out = df.drop_duplicates("order_id")
    logger.info("output_rows=%s", len(out))
    return out

• Registre duração, contagens de linhas e chaves de partição.
• Compare com intervalos históricos para detecção de anomalias.
• Logs JSON estruturados analisam melhor em Loki/ELK.

Relacionado: Melhores Práticas de Engenharia de Dados

Exemplos Intermediários

8. Decorador de Fluxo Prefect

Orquestre tarefas Python com novas tentativas e UI.

from prefect import flow, task
import pandas as pd
 
@task(retries=2, retry_delay_seconds=30)
def extract() -> pd.DataFrame:
    return pd.read_parquet("raw/orders.parquet")
 
@flow(log_prints=True)
def daily_orders():
    df = extract()
    # transform + load ...
 
if __name__ == "__main__":
    daily_orders()

• Novas tentativas de @task isolam falhas transitórias de S3/DB.
• Fluxos Prefect 2.x são Python puro - sem arquivos DAG XML.
• Execute localmente primeiro; o agente implanta em produção.

Relacionado: Prefect & Dagster

9. Ler Conjunto de Dados Particionado

Analise apenas as datas que você precisa.

import pandas as pd
 
df = pd.read_parquet("mart/orders", filters=[("dt", ">=", "2025-01-01")]

• Partições no estilo Hive reduzem I/O em reproduções históricas.
• scan_parquet do Polars oferece o mesmo predicate pushdown.
• Mantenha a coluna de partição no caminho, não duplicada em cada linha, a menos que necessário.

Relacionado: Formatos de Arquivo - layout de partição

Versões da Pilha: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.