Día 3 — El pipeline completo

1Elena Torres  9:31 AM
2Standup — miércoles.
3tú, ¿cómo vas con DATA-147? Hoy era el deadline.
4
5Tú  9:33 AM
6La ingesta está hecha y la limpieza del verano también.
7María me mandó ayer otro CSV (primavera) con formato diferente.
8Estoy adaptando el script con un mapping de columnas como me dijiste.
9¿El deadline sigue siendo hoy o se extiende por el cambio de scope?
10
11Elena Torres  9:35 AM
12El deadline se mantiene para los datos de verano. Lo de primavera lo
13puedes entregar mañana por la mañana. Pero necesito que hoy dejes la
14tabla silver de verano LISTA y con tests básicos. ¿Puedes?
15
16Tú  9:36 AM
17Sí, para las 14:00 debería estar.

Elena es directa: el deadline se mantiene, pero el scope se ajusta

1freshmart-data/
2├── pipelines/
3│   └── campaign_ingestion/
4│       ├── __init__.py
5│       ├── config.py          # Mappings, constantes, enums
6│       ├── clean.py           # Funciones de limpieza
7│       ├── validate.py        # Validaciones de calidad
8│       ├── run.py             # Orquestación del pipeline
9│       └── tests/
10│           ├── __init__.py
11│           ├── test_clean.py
12│           └── test_validate.py
13├── data/
14│   ├── raw/                   # CSVs originales (no se tocan)
15│   └── silver/                # Parquet limpios (output)
16└── README.md

Estructura profesional: cada archivo tiene UNA responsabilidad

1# pipelines/campaign_ingestion/config.py
2"""Configuración centralizada del pipeline de campañas."""
3from enum import Enum
4from pathlib import Path
5
6
7class Campaign(Enum):
8    """Campañas conocidas. Usar este Enum evita typos silenciosos."""
9    VERANO = "verano"
10    PRIMAVERA = "primavera"
11
12
13# Columnas obligatorias después del mapping
14REQUIRED_COLUMNS = [
15    "venta_id", "cliente_id", "producto_id", "fecha_compra",
16    "cantidad", "precio", "descuento_pct", "tienda_id", "campana",
17]
18
19# Mapping de columnas por campaña (origen → destino estándar)
20COLUMN_MAPPINGS: dict[Campaign, dict[str, str]] = {
21    Campaign.VERANO: {},  # Ya viene con nombres correctos
22    Campaign.PRIMAVERA: {
23        "id_transaccion": "venta_id",
24        "id_cliente": "cliente_id",
25        "id_producto": "producto_id",
26        "fecha_venta": "fecha_compra",
27        "unidades": "cantidad",
28        "importe": "precio",
29        "descuento": "descuento_pct",
30        "almacen": "tienda_id",
31        "promocion": "campana",
32    },
33}
34
35# Configuración específica de limpieza por campaña
36CAMPAIGN_CONFIG: dict[Campaign, dict] = {
37    Campaign.VERANO: {
38        "precio_es_string": True,
39        "fechas_mixtas": True,
40        "anonimo_valor": None,  # NULL = anónimo
41    },
42    Campaign.PRIMAVERA: {
43        "precio_es_string": False,
44        "fechas_mixtas": False,
45        "anonimo_valor": -1,  # -1 = anónimo
46    },
47}
48
49# Rutas
50RAW_DIR = Path("data/raw")
51SILVER_DIR = Path("data/silver")
52
53# Rango válido de fechas por campaña
54DATE_RANGES: dict[Campaign, tuple[str, str]] = {
55    Campaign.VERANO: ("2024-06-15", "2024-09-15"),
56    Campaign.PRIMAVERA: ("2024-03-01", "2024-05-31"),
57}

config.py — Todo lo configurable en un solo sitio. Añadir campaña = añadir entrada.

1# pipelines/campaign_ingestion/clean.py
2"""Funciones de limpieza para datos de campañas de Marketing."""
3import logging
4
5import pandas as pd
6
7from .config import Campaign, COLUMN_MAPPINGS, CAMPAIGN_CONFIG, REQUIRED_COLUMNS
8
9logger = logging.getLogger(__name__)
10
11
12def apply_column_mapping(df: pd.DataFrame, campaign: Campaign) -> pd.DataFrame:
13    """Renombra columnas según el mapping de la campaña."""
14    mapping = COLUMN_MAPPINGS[campaign]
15    if mapping:
16        df = df.rename(columns=mapping)
17        logger.info(f"  Mapping aplicado: {len(mapping)} columnas renombradas")
18
19    # Verificar que todas las columnas obligatorias existen
20    missing = set(REQUIRED_COLUMNS) - set(df.columns)
21    if missing:
22        raise ValueError(
23            f"Columnas obligatorias ausentes después del mapping: {missing}"
24        )
25    return df
26
27
28def clean_dates(df: pd.DataFrame, campaign: Campaign) -> pd.DataFrame:
29    """Normaliza la columna fecha_compra a datetime."""
30    config = CAMPAIGN_CONFIG[campaign]
31    try:
32        if config["fechas_mixtas"]:
33            df["fecha_compra"] = pd.to_datetime(
34                df["fecha_compra"], format="mixed", dayfirst=True
35            )
36        else:
37            df["fecha_compra"] = pd.to_datetime(
38                df["fecha_compra"], dayfirst=True
39            )
40    except Exception as e:
41        # No dejar que una fecha rota tumbe el pipeline entero
42        logger.warning(f"  Fechas no parseables encontradas: {e}")
43        df["fecha_compra"] = pd.to_datetime(
44            df["fecha_compra"], format="mixed", dayfirst=True, errors="coerce"
45        )
46        nulas = df["fecha_compra"].isna().sum()
47        logger.warning(f"  {nulas} fechas convertidas a NaT")
48
49    return df
50
51
52def clean_prices(df: pd.DataFrame, campaign: Campaign) -> pd.DataFrame:
53    """Normaliza la columna precio a float."""
54    config = CAMPAIGN_CONFIG[campaign]
55    if config["precio_es_string"]:
56        df["precio"] = (
57            df["precio"]
58            .astype(str)
59            .str.replace("€", "", regex=False)
60            .str.replace(",", ".")
61            .astype(float)
62        )
63        logger.info("  Precios limpiados: eliminado €, coma → punto")
64    else:
65        df["precio"] = df["precio"].astype(float)
66    return df
67
68
69def mark_anonymous(df: pd.DataFrame, campaign: Campaign) -> pd.DataFrame:
70    """Marca ventas anónimas con una columna booleana."""
71    config = CAMPAIGN_CONFIG[campaign]
72    anonimo_valor = config["anonimo_valor"]
73
74    if anonimo_valor is None:
75        df["es_anonimo"] = df["cliente_id"].isna()
76    else:
77        df.loc[df["cliente_id"] == anonimo_valor, "cliente_id"] = None
78        df["es_anonimo"] = df["cliente_id"].isna()
79
80    n_anonimos = df["es_anonimo"].sum()
81    logger.info(f"  Ventas anónimas: {n_anonimos:,}")
82    return df
83
84
85def remove_duplicates(df: pd.DataFrame) -> pd.DataFrame:
86    """Elimina filas 100% duplicadas."""
87    antes = len(df)
88    df = df.drop_duplicates()
89    eliminados = antes - len(df)
90    if eliminados > 0:
91        logger.info(f"  Duplicados eliminados: {eliminados:,}")
92    return df
93
94
95def filter_invalid_rows(df: pd.DataFrame) -> pd.DataFrame:
96    """Filtra filas con cantidad <= 0 (errores de la app)."""
97    antes = len(df)
98    df = df[df["cantidad"] > 0]
99    eliminados = antes - len(df)
100    if eliminados > 0:
101        logger.info(f"  Filas con cantidad <= 0 eliminadas: {eliminados}")
102    return df
103
104
105def run_cleaning(df: pd.DataFrame, campaign: Campaign) -> pd.DataFrame:
106    """Ejecuta el pipeline de limpieza completo."""
107    logger.info(f"Limpiando campaña: {campaign.value} ({len(df):,} filas)")
108
109    df = apply_column_mapping(df, campaign)
110    df = remove_duplicates(df)
111    df = clean_dates(df, campaign)
112    df = clean_prices(df, campaign)
113    df = mark_anonymous(df, campaign)
114    df = filter_invalid_rows(df)
115
116    logger.info(f"  Resultado: {len(df):,} filas limpias")
117    return df

clean.py — Funciones puras, composables y fáciles de testear

1# pipelines/campaign_ingestion/validate.py
2"""Validaciones de calidad para los datos de campaña."""
3import logging
4from dataclasses import dataclass
5
6import pandas as pd
7
8from .config import Campaign, DATE_RANGES
9
10logger = logging.getLogger(__name__)
11
12
13@dataclass
14class ValidationResult:
15    """Resultado de una validación individual."""
16    name: str
17    passed: bool
18    details: str
19
20
21def validate_not_empty(df: pd.DataFrame) -> ValidationResult:
22    """El DataFrame debe tener al menos 1 fila."""
23    passed = len(df) > 0
24    return ValidationResult(
25        name="no_vacio",
26        passed=passed,
27        details=f"{len(df)} filas" if passed else "¡DataFrame vacío!",
28    )
29
30
31def validate_no_null_required_fields(df: pd.DataFrame) -> ValidationResult:
32    """Campos obligatorios no pueden ser NULL (excepto cliente_id en anónimos)."""
33    required = ["venta_id", "producto_id", "fecha_compra", "precio", "cantidad"]
34    nulls = {col: df[col].isna().sum() for col in required}
35    problemas = {k: v for k, v in nulls.items() if v > 0}
36    passed = len(problemas) == 0
37    return ValidationResult(
38        name="campos_obligatorios_no_nulos",
39        passed=passed,
40        details=f"OK" if passed else f"NULLs: {problemas}",
41    )
42
43
44def validate_positive_prices(df: pd.DataFrame) -> ValidationResult:
45    """Todos los precios deben ser positivos."""
46    negativos = (df["precio"] <= 0).sum()
47    passed = negativos == 0
48    return ValidationResult(
49        name="precios_positivos",
50        passed=passed,
51        details="OK" if passed else f"{negativos} precios <= 0",
52    )
53
54
55def validate_dates_in_range(
56    df: pd.DataFrame, campaign: Campaign
57) -> ValidationResult:
58    """Las fechas deben estar dentro del rango de la campaña."""
59    date_min, date_max = DATE_RANGES[campaign]
60    fuera_de_rango = (
61        (df["fecha_compra"] < date_min) | (df["fecha_compra"] > date_max)
62    ).sum()
63    passed = fuera_de_rango == 0
64    return ValidationResult(
65        name="fechas_en_rango",
66        passed=passed,
67        details="OK" if passed else f"{fuera_de_rango} fuera de [{date_min}, {date_max}]",
68    )
69
70
71def validate_no_duplicates(df: pd.DataFrame) -> ValidationResult:
72    """No debe haber filas 100% duplicadas."""
73    dupes = df.duplicated().sum()
74    passed = dupes == 0
75    return ValidationResult(
76        name="sin_duplicados",
77        passed=passed,
78        details="OK" if passed else f"{dupes} filas duplicadas",
79    )
80
81
82def run_validations(
83    df: pd.DataFrame, campaign: Campaign
84) -> list[ValidationResult]:
85    """Ejecuta todas las validaciones y devuelve resultados."""
86    results = [
87        validate_not_empty(df),
88        validate_no_null_required_fields(df),
89        validate_positive_prices(df),
90        validate_dates_in_range(df, campaign),
91        validate_no_duplicates(df),
92    ]
93
94    # Logging
95    passed = sum(1 for r in results if r.passed)
96    total = len(results)
97    logger.info(f"Validaciones: {passed}/{total} pasaron")
98    for r in results:
99        icon = "✓" if r.passed else "✗"
100        logger.info(f"  {icon} {r.name}: {r.details}")
101
102    return results

validate.py — Validaciones claras, ejecutables y con resultados estructurados

1# pipelines/campaign_ingestion/tests/test_validate.py
2"""Tests unitarios para las funciones de validación."""
3import pandas as pd
4import pytest
5
6from campaign_ingestion.config import Campaign
7from campaign_ingestion.validate import (
8    validate_not_empty,
9    validate_no_null_required_fields,
10    validate_positive_prices,
11    validate_dates_in_range,
12    validate_no_duplicates,
13)
14
15
16@pytest.fixture
17def df_valido():
18    """DataFrame que pasa TODAS las validaciones."""
19    return pd.DataFrame({
20        "venta_id": [1, 2, 3],
21        "cliente_id": [100, None, 200],  # None es anónimo, OK
22        "producto_id": [2001, 2002, 2003],
23        "fecha_compra": pd.to_datetime(["2024-07-01", "2024-07-15", "2024-08-01"]),
24        "cantidad": [1, 2, 3],
25        "precio": [2.45, 8.99, 1.20],
26        "descuento_pct": [0.0, 10.0, 5.0],
27        "tienda_id": [1, 2, 1],
28        "campana": ["verano", "verano", "verano"],
29        "es_anonimo": [False, True, False],
30    })
31
32
33@pytest.fixture
34def df_vacio():
35    """DataFrame sin filas."""
36    return pd.DataFrame(columns=["venta_id", "producto_id", "fecha_compra", "precio", "cantidad"])
37
38
39class TestValidateNotEmpty:
40    def test_pasa_con_datos(self, df_valido):
41        result = validate_not_empty(df_valido)
42        assert result.passed is True
43
44    def test_falla_sin_datos(self, df_vacio):
45        result = validate_not_empty(df_vacio)
46        assert result.passed is False
47
48
49class TestValidatePositivePrices:
50    def test_pasa_con_precios_positivos(self, df_valido):
51        result = validate_positive_prices(df_valido)
52        assert result.passed is True
53
54    def test_falla_con_precio_negativo(self, df_valido):
55        df_valido.loc[0, "precio"] = -5.0
56        result = validate_positive_prices(df_valido)
57        assert result.passed is False
58        assert "-5" in result.details or "1" in result.details
59
60
61class TestValidateDatesInRange:
62    def test_pasa_con_fechas_en_rango_verano(self, df_valido):
63        result = validate_dates_in_range(df_valido, Campaign.VERANO)
64        assert result.passed is True
65
66    def test_falla_con_fecha_fuera_de_rango(self, df_valido):
67        df_valido.loc[0, "fecha_compra"] = pd.Timestamp("2024-01-01")
68        result = validate_dates_in_range(df_valido, Campaign.VERANO)
69        assert result.passed is False
70
71
72class TestValidateNoDuplicates:
73    def test_pasa_sin_duplicados(self, df_valido):
74        result = validate_no_duplicates(df_valido)
75        assert result.passed is True
76
77    def test_falla_con_duplicados(self, df_valido):
78        df_con_dupes = pd.concat([df_valido, df_valido.iloc[[0]]])
79        result = validate_no_duplicates(df_con_dupes)
80        assert result.passed is False

test_validate.py — Tests que verifican que tus validaciones funcionan

1# 🟦 Windows (PowerShell) / 🍎 Mac (Terminal)
2git checkout -b feat/DATA-147-campaign-ingestion
3git add pipelines/campaign_ingestion/
4git commit -m "feat(DATA-147): pipeline de ingesta para campañas de Marketing
5
6- Pipeline raw → silver para campañas de Marketing
7- Mapping configurable para diferentes formatos (verano, primavera)
8- Funciones de limpieza: fechas, precios, duplicados, anónimos
9- Validaciones de calidad con resultados estructurados
10- Tests unitarios con pytest"
11git push -u origin feat/DATA-147-campaign-ingestion

Branch descriptiva, commit con contexto, push con tracking

1╔══════════════════════════════════════════════════════════════════╗
2║  PR #234: feat: campaign ingestion pipeline (DATA-147)           ║
3╠══════════════════════════════════════════════════════════════════╣
4║                                                                  ║
5║  ## Qué hace                                                     ║
6║  Pipeline de ingesta para campañas de Marketing que:             ║
7║  - Lee CSVs de raw (originales sin tocar)                        ║
8║  - Aplica mapping de columnas según la campaña                   ║
9║  - Ejecuta limpieza (fechas, precios, duplicados, anónimos)      ║
10║  - Valida calidad de los datos resultantes                       ║
11║  - Escribe el resultado en silver como Parquet                   ║
12║                                                                  ║
13║  ## Archivos nuevos                                              ║
14║  pipelines/campaign_ingestion/ (6 archivos, ~280 líneas)         ║
15║                                                                  ║
16║  ## Cómo probar                                                  ║
17║  cd pipelines/campaign_ingestion                                 ║
18║  python run.py --campaign verano                                 ║
19║  pytest tests/ -v                                                ║
20║                                                                  ║
21║  ## Decisiones de diseño                                         ║
22║  - Enum para campañas (evita typos silenciosos)                  ║
23║  - Funciones de limpieza puras (sin side effects)                ║
24║  - Validaciones como dataclass (resultado estructurado)          ║
25║  - Ventas anónimas se MARCAN, no se eliminan                     ║
26║                                                                  ║
27║  ## Contexto                                                     ║
28║  María necesita estos datos para la presentación del jueves.     ║
29║  Los datos de primavera se añadirán mañana (mismo pipeline).     ║
30║                                                                  ║
31╚══════════════════════════════════════════════════════════════════╝

Una buena PR tiene: qué, cómo probar, decisiones de diseño y contexto de negocio

1# ──────────────────────────────────────────────────────────────
2# COMENTARIO 1 de Elena en clean.py, línea 42:
3# ──────────────────────────────────────────────────────────────
4# Tu código original:
5df["fecha_compra"] = pd.to_datetime(
6    df["fecha_compra"], infer_datetime_format=True
7)
8
9# Elena Torres comentó:
10# ❌ No uses infer_datetime_format — está deprecated desde Pandas 2.0.
11# Usa format='mixed' con dayfirst=True. Y pon un try/except por si
12# hay fechas completamente inválidas — no quiero que el pipeline
13# entero falle por UNA fecha rota. Usa errors='coerce' como fallback.

Comentario 1: deprecated API + falta de manejo de errores

1# ──────────────────────────────────────────────────────────────
2# COMENTARIO 2 de Elena en run.py, línea 8:
3# ──────────────────────────────────────────────────────────────
4# Tu código original:
5INPUT_PATH = "data/raw/marketing/campana_verano_ventas/2024/09/16/"
6OUTPUT_PATH = "data/silver/campaign_ventas.parquet"
7
8# Elena Torres comentó:
9# ⚠️ El path está hardcodeado. En producción el bucket no se llama
10# igual que en local. Usa variables de entorno o argumentos de CLI:
11#
12# import os
13# INPUT_PATH = os.environ.get("RAW_PATH", "data/raw/")
14# OUTPUT_PATH = os.environ.get("SILVER_PATH", "data/silver/")
15#
16# O mejor: usa argparse para recibir --input y --output como args.

Comentario 2: paths hardcodeados = bomba de tiempo en producción

1# ──────────────────────────────────────────────────────────────
2# COMENTARIO 3 de Elena en validate.py, línea 31:
3# ──────────────────────────────────────────────────────────────
4# Elena Torres comentó:
5# 👍 Bien las validaciones. Pero falta una CRÍTICA:
6# ¿Qué pasa si el CSV viene vacío? Añade un check de que el
7# DataFrame tiene al menos 1 fila ANTES de las demás validaciones.
8# Te sorprendería cuántas veces un archivo llega vacío por un
9# error en el sistema de origen y nadie se entera durante días.

Comentario 3: el caso edge que parece obvio pero que te puede morder

1# ──────────────────────────────────────────────────────────────
2# COMENTARIO 4 de Elena en config.py, línea 2:
3# ──────────────────────────────────────────────────────────────
4# Tu código original:
5COLUMN_MAPPINGS = {
6    'verano': {},
7    'primavera': {...},
8}
9
10# Elena Torres comentó:
11# Nit: los nombres de campaña como keys del dict deberían ser un
12# Enum o al menos constantes. Si alguien escribe 'Verano' en vez
13# de 'verano' explota silenciosamente (el dict devuelve el valor
14# por defecto sin avisar). Con un Enum te da un error claro.
15#
16# from enum import Enum
17# class Campaign(Enum):
18#     VERANO = "verano"
19#     PRIMAVERA = "primavera"

Comentario 4: strings mágicas → Enum. Explícito mejor que implícito.

1# Corrección 1: usar format='mixed' + try/except
2git commit -m "fix: usar pd.to_datetime format=mixed (deprecated infer_datetime_format)
3
4Ref: review comment de Elena — infer_datetime_format está deprecated
5desde Pandas 2.0. Añadido try/except con errors='coerce' como fallback."
6
7# Corrección 2: parametrizar paths
8git commit -m "refactor: paths como argumentos de CLI en vez de hardcoded
9
10Añadido argparse con --input y --output. Default apunta a local
11para desarrollo pero permite override en producción."
12
13# Corrección 3: validación de DataFrame vacío
14git commit -m "feat: añadir validate_not_empty como primera validación
15
16Check de que el DataFrame tiene ≥1 fila antes de ejecutar las
17demás validaciones. Evita procesar archivos vacíos en silencio."
18
19# Corrección 4: Enum para campañas
20git commit -m "refactor: Campaign Enum en vez de strings como keys
21
22Strings mágicas → Enum. Typos dan error explícito en vez de
23devolver None silenciosamente."
24
25# Push de todas las correcciones
26git push

Cada corrección en un commit separado — facilita el re-review

1Elena Torres  16:32 PM
2PR aprobada ✅
3Buen primer pipeline. Mergea cuando quieras.
4Mañana quiero que hagas lo mismo con primavera y que construyas
5la tabla gold unificada (verano + primavera juntas).

Aprobada. "Buen primer pipeline" de Elena vale su peso en oro.

1Elena: ¿Qué tal la semana? ¿Algún bloqueante?
2
3Tú: Bien, la verdad. Lo de María cambiando requisitos me pilló un poco
4    desprevenido...
5
6Elena: Bienvenido a la vida real. María siempre hace eso. El truco es:
7       1. No digas que sí a todo sin preguntar primero
8       2. Comunica el impacto en tiempo ("puedo hacerlo pero será 1 día más")
9       3. Pon por escrito qué se acordó (Slack es tu amigo)
10
11Elena: Por lo demás, buen trabajo. Tu código es limpio para ser tu primera
12       semana. Solo te falta rodaje con el stack. Mañana te voy a pedir
13       algo más complejo — la tabla gold para Marketing.
14
15Tú: ¿Algo que debería leer o estudiar para prepararme?
16
17Elena: Lee cómo funcionan las agregaciones en Pandas y cómo se escribe a
18       Parquet con particionado. Si no has usado groupby con múltiples
19       funciones de agregación, repásalo esta noche. Mañana lo vas a necesitar.

El 1:1 es tu espacio para aprender sin presión del equipo