Fundamentos de Ingeniería de Datos

9 ejemplos para empezar con Ingeniería de Datos - 7 básicos y 2 intermedios.

Prerrequisitos

• Python 3.14.0 con pandas 2.2+, pyarrow y familiaridad con la orquestación.
• Configuración rápida: uv pip install pandas pyarrow prefect

Ejemplos Básicos

1. Extracción a Parquet

Almacena datos sin procesar en almacenamiento columnar antes de las transformaciones.

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 incrusta el esquema y comprime mejor que CSV.
• Las rutas de partición como dt=2025-01-15/ aceleran las lecturas selectivas.
• Nunca sobrescribas datos sin procesar; anexa particiones por fecha de ejecución.

Relacionado: Formatos de Archivo - Parquet y Arrow

2. Transformación con pandas

Limpia y da forma a los datos en un paso scriptable.

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 etapa escribe un artefacto que puedes auditar.
• Registra el número de filas de entrada/salida en cada paso.
• Mantén las reglas de negocio en Python versionado.

Relacionado: Limpieza y Transformación de Datos

3. Carga Idempotente

Actualiza o sobrescribe particiones para que los reintentos no dupliquen.

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)  # sobrescribe la partición para esta fecha

• La clave de partición limita la idempotencia a un día.
• Volver a ejecutar la misma run_dt reemplaza, no anexa duplicados.
• Documenta la política de datos que llegan tarde por separado.

Relacionado: Fiabilidad del Flujo de Trabajo - decisiones de idempotencia

4. Programación con cron

Los trabajos por lotes se ejecutan en un reloj cuando la tolerancia a la latencia es 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 está bien para canalizaciones MVP de un solo servidor.
• Redirige los registros a archivos o a un sistema de registro centralizado.
• Sal con un código distinto de cero en caso de error para que la monitorización alerte.

Relacionado: Airflow - programación de DAGs

5. Límites de Tareas

Una función por paso con entradas/salidas 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)

• Las funciones puras simplifican las pruebas unitarias.
• Pasa rutas o configuraciones; evita globales ocultos.
• El orquestador conecta extract -> transform -> load.

Relacionado: Prefect & Dagster

6. Validación de Esquema

Rechaza datos incorrectos antes de que lleguen a los 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)

• Los fallos de validación deben bloquear las tareas posteriores.
• Almacena datos en cuarentena en Parquet para las filas rechazadas.
• Versiona los esquemas junto con el código de la canalización.

Relacionado: Validación y Calidad de Datos

7. Registro de Recuentos de Filas

La visibilidad operativa comienza con 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

• Registra la duración, los recuentos de filas y las claves de partición.
• Compara con rangos históricos para la detección de anomalías.
• Los registros JSON estructurados se analizan mejor en Loki/ELK.

Relacionado: Mejores Prácticas de Ingeniería de Datos

Ejemplos Intermedios

8. Decorador de Flujo de Prefect

Orquesta tareas de Python con reintentos e interfaz de usuario.

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()

• Los reintentos de @task aíslan fallos transitorios de S3/DB.
• Los flujos de Prefect 2.x son Python plano; sin archivos DAG XML.
• Ejecuta localmente primero; el agente lo implementa en producción.

Relacionado: Prefect & Dagster

9. Lectura de Conjunto de Datos Particionado

Escanea solo las fechas que necesitas.

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

• Las particiones estilo Hive reducen la E/S en reproducciones históricas.
• scan_parquet de Polars ofrece la misma descarga de predicados.
• Mantén la columna de partición en la ruta, no duplicada en cada fila a menos que sea necesario.

Relacionado: Formatos de Archivo - diseño de partición

Versiones de la Pila: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+, y uv 0.6+.