Día 2 -- Pipeline ETL: PostgreSQL a S3 a DuckDB

1streamvibe-data/
2├── etl/
3│   ├── __init__.py
4│   ├── config.py          # Configuración (env vars, paths, constantes)
5│   ├── extract.py         # Lectura de PostgreSQL réplica
6│   ├── transform.py       # Limpieza, dedup, enriquecimiento
7│   ├── load.py            # Escritura a S3 + refresh DuckDB
8│   └── pipeline.py        # Orquestador: extract -> transform -> load
9├── analytics/
10│   ├── __init__.py
11│   ├── metrics.py         # Cálculo de DAU, MAU, retention, etc.
12│   └── content.py         # Análisis de rendimiento de contenido
13├── utils/
14│   ├── __init__.py
15│   ├── logging_config.py  # Setup de logging estructurado
16│   └── watermark.py       # Gestión de watermarks incrementales
17├── tests/
18│   ├── test_extract.py
19│   ├── test_transform.py
20│   └── test_metrics.py
21├── cron/
22│   └── crontab.txt        # Definición de jobs cron
23├── requirements.txt
24├── .env.example
25└── README.md

Estructura limpia y mantenible. Un senior organiza su código desde el día 1.

1# etl/config.py
2import os
3from dataclasses import dataclass
4
5@dataclass(frozen=True)
6class Config:
7    """Configuración centralizada del pipeline."""
8    # PostgreSQL réplica (solo lectura)
9    pg_host: str = os.getenv("PG_REPLICA_HOST", "replica.streamvibe.internal")
10    pg_port: int = int(os.getenv("PG_REPLICA_PORT", "5432"))
11    pg_database: str = os.getenv("PG_DATABASE", "streamvibe")
12    pg_user: str = os.getenv("PG_USER", "data_reader")
13    pg_password: str = os.getenv("PG_PASSWORD", "")
14
15    # S3
16    s3_bucket: str = os.getenv("S3_BUCKET", "streamvibe-data-lake")
17    s3_region: str = os.getenv("AWS_REGION", "eu-west-1")
18
19    # Paths
20    raw_prefix: str = "raw"
21    silver_prefix: str = "silver"
22    gold_prefix: str = "gold"
23    watermark_key: str = "metadata/watermarks.json"
24
25    # DuckDB
26    duckdb_path: str = os.getenv("DUCKDB_PATH", "/data/streamvibe.duckdb")
27
28    @property
29    def pg_connection_string(self) -> str:
30        return (
31            f"host={self.pg_host} port={self.pg_port} "
32            f"dbname={self.pg_database} user={self.pg_user} "
33            f"password={self.pg_password}"
34        )
35
36config = Config()

Configuración via environment variables -- patrón 12-factor apps.

1# etl/extract.py
2import psycopg2
3import pandas as pd
4import logging
5from datetime import datetime, timezone
6from .config import config
7
8logger = logging.getLogger(__name__)
9
10
11def get_watermark(s3_client, table_name: str) -> datetime:
12    """Lee el último timestamp procesado para una tabla."""
13    import json
14    try:
15        obj = s3_client.get_object(
16            Bucket=config.s3_bucket,
17            Key=config.watermark_key
18        )
19        watermarks = json.loads(obj['Body'].read())
20        ts = watermarks.get(table_name)
21        if ts:
22            return datetime.fromisoformat(ts)
23    except s3_client.exceptions.NoSuchKey:
24        pass
25    # Primera ejecución: todo el histórico
26    return datetime(2023, 1, 1, tzinfo=timezone.utc)
27
28
29def extract_playbacks(since: datetime) -> pd.DataFrame:
30    """Extrae reproducciones nuevas desde el watermark."""
31    query = """
32        SELECT
33            playback_id,
34            user_id,
35            content_id,
36            started_at,
37            ended_at,
38            watch_seconds,
39            completed,
40            created_at
41        FROM playbacks
42        WHERE created_at > %(since)s
43        ORDER BY created_at ASC
44        LIMIT 500000
45    """
46    logger.info(f"Extrayendo playbacks desde {since.isoformat()}")
47    with psycopg2.connect(config.pg_connection_string) as conn:
48        df = pd.read_sql(query, conn, params={"since": since})
49
50    logger.info(f"Extraídas {len(df):,} filas de playbacks")
51    return df
52
53
54def extract_users() -> pd.DataFrame:
55    """Extrae tabla completa de usuarios (dimensión, ~42K filas)."""
56    query = """
57        SELECT user_id, email, username, plan, country,
58               registered_at, last_login_at, device_preference
59        FROM users
60    """
61    with psycopg2.connect(config.pg_connection_string) as conn:
62        df = pd.read_sql(query, conn)
63    logger.info(f"Extraídos {len(df):,} usuarios")
64    return df
65
66
67def extract_content() -> pd.DataFrame:
68    """Extrae catálogo completo de contenido (dimensión, ~1200 filas)."""
69    query = """
70        SELECT content_id, title, content_type, genre,
71               duration_seconds, release_year, director,
72               rating_imdb, license_cost_eur, added_at, is_premium
73        FROM content
74    """
75    with psycopg2.connect(config.pg_connection_string) as conn:
76        df = pd.read_sql(query, conn)
77    logger.info(f"Extraídos {len(df):,} títulos de contenido")
78    return df

Extracción incremental con LIMIT para no saturar la réplica. Patrón defensivo.

1# etl/transform.py
2import pandas as pd
3import numpy as np
4import logging
5
6logger = logging.getLogger(__name__)
7
8
9def transform_playbacks(df: pd.DataFrame) -> pd.DataFrame:
10    """Transforma playbacks raw en formato silver limpio."""
11    if df.empty:
12        return df
13
14    initial_count = len(df)
15
16    # 1. Deduplicar por playback_id
17    df = df.drop_duplicates(subset=['playback_id'], keep='first')
18
19    # 2. Parsear timestamps con timezone
20    df['started_at'] = pd.to_datetime(df['started_at'], utc=True)
21    df['ended_at'] = pd.to_datetime(df['ended_at'], utc=True, errors='coerce')
22
23    # 3. Validar watch_seconds (no negativo, no mayor que 24h)
24    df['watch_seconds'] = pd.to_numeric(
25        df['watch_seconds'], errors='coerce'
26    ).fillna(0).astype(int)
27    df = df[(df['watch_seconds'] >= 0) & (df['watch_seconds'] <= 86400)]
28
29    # 4. Calcular campos derivados
30    df['event_date'] = df['started_at'].dt.date
31    df['event_hour'] = df['started_at'].dt.hour
32    df['day_of_week'] = df['started_at'].dt.dayofweek  # 0=lunes
33    df['is_weekend'] = df['day_of_week'].isin([5, 6])
34
35    # 5. Categorizar duración de sesión
36    df['session_category'] = pd.cut(
37        df['watch_seconds'],
38        bins=[0, 300, 1200, 3600, 7200, 86400],
39        labels=['< 5 min', '5-20 min', '20-60 min', '1-2 horas', '> 2 horas']
40    )
41
42    # 6. Flag de sesión significativa (> 2 min de watch)
43    df['is_meaningful'] = df['watch_seconds'] > 120
44
45    final_count = len(df)
46    dropped = initial_count - final_count
47    if dropped > 0:
48        logger.warning(f"Descartadas {dropped} filas inválidas ({dropped/initial_count*100:.1f}%)")
49
50    logger.info(f"Transformación completada: {final_count:,} filas válidas")
51    return df.reset_index(drop=True)
52
53
54def transform_users(df: pd.DataFrame) -> pd.DataFrame:
55    """Limpia y enriquece la dimensión de usuarios."""
56    df['registered_at'] = pd.to_datetime(df['registered_at'], utc=True)
57    df['last_login_at'] = pd.to_datetime(df['last_login_at'], utc=True, errors='coerce')
58
59    # Calcular antigüedad en días
60    now = pd.Timestamp.now(tz='UTC')
61    df['days_since_registration'] = (now - df['registered_at']).dt.days
62
63    # Cohorte mensual de registro
64    df['registration_cohort'] = df['registered_at'].dt.to_period('M').astype(str)
65
66    # Estandarizar plan
67    df['plan'] = df['plan'].fillna('free').str.lower().str.strip()
68
69    return df

Transformación defensiva: dedup, validación, campos derivados para analytics.

1# etl/load.py
2import boto3
3import duckdb
4import pandas as pd
5import pyarrow as pa
6import pyarrow.parquet as pq
7import logging
8from io import BytesIO
9from .config import config
10
11logger = logging.getLogger(__name__)
12
13
14def write_parquet_to_s3(
15    df: pd.DataFrame,
16    s3_client,
17    prefix: str,
18    partition_col: str = None
19):
20    """Escribe un DataFrame como Parquet en S3, opcionalmente particionado."""
21    if df.empty:
22        logger.warning(f"DataFrame vacío, no se escribe en {prefix}")
23        return
24
25    if partition_col and partition_col in df.columns:
26        # Escribir una partición por valor único
27        for value, partition_df in df.groupby(partition_col):
28            key = f"{prefix}/{partition_col}={value}/data.parquet"
29            _write_single_parquet(partition_df, s3_client, key)
30    else:
31        key = f"{prefix}/data.parquet"
32        _write_single_parquet(df, s3_client, key)
33
34
35def _write_single_parquet(df: pd.DataFrame, s3_client, key: str):
36    """Escribe un único archivo Parquet a S3."""
37    table = pa.Table.from_pandas(df)
38    buffer = BytesIO()
39    pq.write_table(table, buffer, compression='snappy')
40    buffer.seek(0)
41
42    s3_client.put_object(
43        Bucket=config.s3_bucket,
44        Key=key,
45        Body=buffer.getvalue(),
46        ContentType='application/octet-stream'
47    )
48    size_mb = buffer.getbuffer().nbytes / (1024 * 1024)
49    logger.info(f"Escrito s3://{config.s3_bucket}/{key} ({size_mb:.2f} MB)")
50
51
52def setup_duckdb_analytics():
53    """Configura DuckDB para leer del lake en S3."""
54    con = duckdb.connect(config.duckdb_path)
55
56    # Instalar y cargar extensión para S3
57    con.execute("INSTALL httpfs; LOAD httpfs;")
58    con.execute(f"SET s3_region = '{config.s3_region}';")
59
60    # Crear vistas sobre los archivos Parquet en S3
61    bucket = config.s3_bucket
62    con.execute(f"""
63        CREATE OR REPLACE VIEW silver_playbacks AS
64        SELECT * FROM read_parquet(
65            's3://{bucket}/silver/playbacks/event_date=*/data.parquet',
66            hive_partitioning = true
67        );
68    """)
69    con.execute(f"""
70        CREATE OR REPLACE VIEW dim_users AS
71        SELECT * FROM read_parquet('s3://{bucket}/silver/users/data.parquet');
72    """)
73    con.execute(f"""
74        CREATE OR REPLACE VIEW dim_content AS
75        SELECT * FROM read_parquet('s3://{bucket}/silver/content/data.parquet');
76    """)
77
78    # Verificar que funciona
79    count = con.execute("SELECT COUNT(*) FROM silver_playbacks").fetchone()[0]
80    logger.info(f"DuckDB configurado. silver_playbacks: {count:,} filas")
81
82    con.close()
83    return count

DuckDB lee Parquet de S3 directamente via vistas -- como un Athena gratuito y local.

1# etl/pipeline.py
2import boto3
3import json
4import logging
5from datetime import datetime, timezone
6from .config import config
7from .extract import extract_playbacks, extract_users, extract_content, get_watermark
8from .transform import transform_playbacks, transform_users
9from .load import write_parquet_to_s3, setup_duckdb_analytics
10
11logger = logging.getLogger(__name__)
12
13
14def run_daily_pipeline():
15    """Pipeline diario completo: extract -> transform -> load."""
16    start_time = datetime.now(timezone.utc)
17    logger.info(f"=== Pipeline iniciado: {start_time.isoformat()} ===")
18
19    s3 = boto3.client('s3', region_name=config.s3_region)
20
21    try:
22        # 1. EXTRACT
23        watermark = get_watermark(s3, 'playbacks')
24        logger.info(f"Watermark playbacks: {watermark.isoformat()}")
25
26        playbacks_raw = extract_playbacks(since=watermark)
27        users_raw = extract_users()
28        content_raw = extract_content()
29
30        if playbacks_raw.empty:
31            logger.info("No hay nuevos playbacks. Pipeline finalizado.")
32            return {"status": "no_new_data", "duration_seconds": 0}
33
34        # 2. TRANSFORM
35        playbacks_clean = transform_playbacks(playbacks_raw)
36        users_clean = transform_users(users_raw)
37
38        # 3. LOAD a S3
39        write_parquet_to_s3(
40            playbacks_clean, s3,
41            prefix=f"{config.silver_prefix}/playbacks",
42            partition_col='event_date'
43        )
44        write_parquet_to_s3(users_clean, s3, prefix=f"{config.silver_prefix}/users")
45        write_parquet_to_s3(content_raw, s3, prefix=f"{config.silver_prefix}/content")
46
47        # 4. Actualizar watermark
48        new_watermark = playbacks_clean['created_at'].max().isoformat()
49        _update_watermark(s3, 'playbacks', new_watermark)
50
51        # 5. Refresh DuckDB
52        total_rows = setup_duckdb_analytics()
53
54        duration = (datetime.now(timezone.utc) - start_time).total_seconds()
55        result = {
56            "status": "success",
57            "new_playbacks": len(playbacks_clean),
58            "total_rows_in_lake": total_rows,
59            "duration_seconds": round(duration, 1),
60        }
61        logger.info(f"=== Pipeline completado: {result} ===")
62        return result
63
64    except Exception as e:
65        logger.error(f"Pipeline FALLIDO: {e}", exc_info=True)
66        return {"status": "error", "error": str(e)}
67
68
69def _update_watermark(s3_client, table_name: str, timestamp_iso: str):
70    """Actualiza el watermark en S3."""
71    try:
72        obj = s3_client.get_object(Bucket=config.s3_bucket, Key=config.watermark_key)
73        watermarks = json.loads(obj['Body'].read())
74    except s3_client.exceptions.NoSuchKey:
75        watermarks = {}
76
77    watermarks[table_name] = timestamp_iso
78    s3_client.put_object(
79        Bucket=config.s3_bucket,
80        Key=config.watermark_key,
81        Body=json.dumps(watermarks, indent=2).encode(),
82    )

Pipeline completo con logging, watermarks y error handling. Listo para cron.

1# cron/crontab.txt
2# Pipeline diario: extrae datos nuevos a las 03:00 (mínimo tráfico en la réplica)
30 3 * * * cd /opt/streamvibe-data && /opt/streamvibe-data/venv/bin/python -m etl.pipeline >> /var/log/streamvibe-etl.log 2>&1
4
5# Recalcular métricas gold: a las 03:30 (después de que el pipeline silver termine)
630 3 * * * cd /opt/streamvibe-data && /opt/streamvibe-data/venv/bin/python -m analytics.metrics >> /var/log/streamvibe-metrics.log 2>&1
7
8# Backup del DuckDB: a las 04:00
90 4 * * * cp /data/streamvibe.duckdb /data/backups/streamvibe_$(date +\%Y\%m\%d).duckdb

Tres jobs cron: ETL a las 3, métricas a las 3:30, backup a las 4. Simple y robusto.

1# tests/test_transform.py
2import pandas as pd
3import pytest
4from etl.transform import transform_playbacks
5
6
7def test_deduplication():
8    """Los playbacks duplicados se eliminan."""
9    df = pd.DataFrame({
10        'playback_id': [1, 1, 2, 3],
11        'user_id': [10, 10, 20, 30],
12        'content_id': [100, 100, 200, 300],
13        'started_at': ['2024-03-15'] * 4,
14        'watch_seconds': [100, 100, 200, 300],
15        'completed': [False, False, True, False],
16        'created_at': ['2024-03-15'] * 4,
17    })
18    result = transform_playbacks(df)
19    assert len(result) == 3  # Uno de los duplicados eliminado
20
21
22def test_invalid_watch_seconds_filtered():
23    """watch_seconds negativos o > 24h se descartan."""
24    df = pd.DataFrame({
25        'playback_id': [1, 2, 3],
26        'user_id': [10, 20, 30],
27        'content_id': [100, 200, 300],
28        'started_at': ['2024-03-15'] * 3,
29        'watch_seconds': [-5, 90000, 1000],  # Negativo, > 24h, válido
30        'completed': [False, False, True],
31        'created_at': ['2024-03-15'] * 3,
32    })
33    result = transform_playbacks(df)
34    assert len(result) == 1  # Solo el de 1000s es válido
35    assert result.iloc[0]['watch_seconds'] == 1000
36
37
38def test_null_user_filtered():
39    """Registros sin user_id se descartan."""
40    df = pd.DataFrame({
41        'playback_id': [1, 2],
42        'user_id': [None, 20],
43        'content_id': [100, 200],
44        'started_at': ['2024-03-15'] * 2,
45        'watch_seconds': [100, 200],
46        'completed': [False, True],
47        'created_at': ['2024-03-15'] * 2,
48    })
49    result = transform_playbacks(df)
50    assert len(result) == 1
51    assert result.iloc[0]['user_id'] == 20
52
53
54def test_empty_dataframe():
55    """Un DataFrame vacío retorna vacío sin error."""
56    df = pd.DataFrame()
57    result = transform_playbacks(df)
58    assert len(result) == 0

Tests básicos pero esenciales. Si algo rompe la transformación, te enteras antes de producción.