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
29LIMITE_LOTE = 500_000
30
31
32def extract_playbacks(since: datetime) -> pd.DataFrame:
33    """Extrae reproducciones nuevas desde el watermark."""
34    query = """
35        SELECT
36            playback_id,
37            user_id,
38            content_id,
39            started_at,
40            ended_at,
41            watch_seconds,
42            completed,
43            created_at
44        FROM playbacks
45        WHERE created_at > %(since)s
46        ORDER BY created_at ASC
47        LIMIT %(limite)s
48    """
49    logger.info(f"Extrayendo playbacks desde {since.isoformat()}")
50    with psycopg2.connect(config.pg_connection_string) as conn:
51        df = pd.read_sql(query, conn, params={"since": since, "limite": LIMITE_LOTE})
52
53    # EL BORDE DEL LOTE: aqui se pierden filas sin que nadie se entere.
54    #
55    # El LIMIT corta por numero de filas, no por instante. Si en el segundo
56    # 03:00:07 se insertaron 40 reproducciones y el corte cae en la numero 12,
57    # el lote se lleva 12 y deja 28 dentro de la base. Pero la marca de agua
58    # avanza a 03:00:07 (es el max del lote), y manana la consulta pide
59    # created_at > 03:00:07. Esas 28 filas no las va a pedir nadie nunca mas.
60    # No hay error, no hay aviso: simplemente no estan.
61    #
62    # El instinto es cambiar el > por un >=, y con esta carga es peor: el
63    # fichero se nombra por el rango del lote, asi que las 12 filas del solape
64    # se escribirian otra vez en un fichero nuevo de la misma particion. Se
65    # cambia perder filas por contarlas dos veces, que encima se ve menos.
66    #
67    # La solucion es no partir un instante en dos: si el lote llego al tope,
68    # se suelta el ultimo instante ENTERO y se lo lleva el lote siguiente
69    # completo. Se procesan menos filas hoy y ninguna se queda a medias.
70    if len(df) == LIMITE_LOTE:
71        corte = df['created_at'].max()
72        df = df[df['created_at'] < corte]
73        logger.warning(
74            f"Lote al tope ({LIMITE_LOTE:,}): se deja el instante {corte} "
75            f"para la proxima ejecucion. Quedan {len(df):,} filas."
76        )
77        # Caso limite: si TODAS las filas del lote comparten el mismo instante,
78        # esto vacia el DataFrame y el pipeline no avanzaria nunca. Con 500.000
79        # filas en un segundo no pasa aqui, pero el dia que pase hay que verlo
80        # en el log y no en la metrica de Marta.
81        if df.empty:
82            raise RuntimeError(
83                f"{LIMITE_LOTE:,} filas con el mismo created_at ({corte}): "
84                "sube LIMITE_LOTE o trocea la extraccion por rango de fechas."
85            )
86
87    logger.info(f"Extraídas {len(df):,} filas de playbacks")
88    return df
89
90
91def extract_users() -> pd.DataFrame:
92    """Extrae tabla completa de usuarios (dimensión, ~20K filas)."""
93    query = """
94        SELECT user_id, email, username, plan, country,
95               registered_at, last_login_at, device_preference
96        FROM users
97    """
98    with psycopg2.connect(config.pg_connection_string) as conn:
99        df = pd.read_sql(query, conn)
100    logger.info(f"Extraídos {len(df):,} usuarios")
101    return df
102
103
104def extract_content() -> pd.DataFrame:
105    """Extrae catálogo completo de contenido (dimensión, ~1200 filas)."""
106    query = """
107        SELECT content_id, title, content_type, genre,
108               duration_seconds, release_year, director,
109               rating_imdb, license_cost_eur, added_at, is_premium
110        FROM content
111    """
112    # OJO CON LA UNIDAD. En el fichero del caso la columna se llama
113    # duration_MINUTES, no duration_seconds. Si copias esta consulta contra el
114    # fichero sin darte cuenta, cualquier calculo que compare el tiempo visto
115    # (segundos) con la duracion del titulo sale multiplicado o dividido por 60 --
116    # y eso es exactamente la tasa de finalizacion que calculas el miercoles por
117    # la tarde para Sofia. Comprueba la unidad ANTES de dividir, siempre.
118    with psycopg2.connect(config.pg_connection_string) as conn:
119        df = pd.read_sql(query, conn)
120    logger.info(f"Extraídos {len(df):,} títulos de contenido")
121    return df

Extracción incremental con tope de lote y corte por instante completo.

1# etl/transform.py
2import pandas as pd
3import numpy as np
4import logging
5
6logger = logging.getLogger(__name__)
7
8# Cuanto hay que ver para que cuente como una vision y no como un clic.
9#
10# Va aqui, en un solo sitio, porque manana el DAU se calcula filtrando por la
11# columna que sale de esta constante. Un umbral repetido a mano en cada consulta
12# es un umbral que dentro de dos meses vale 30 en unas y 120 en otras, y entonces
13# el mismo DAU da dos numeros segun quien lo calcule.
14#
15# Y el numero no se elige por intuicion: se mira cuanta gente cae en la frontera.
16# En el extracto de este caso, entre los 30 y los 120 segundos hay 79 sesiones de
17# 18.492, asi que aqui da igual cual sea. Saber que da igual vale mas que acertar
18# por suerte.
19UMBRAL_SESION_SIGNIFICATIVA = 120  # segundos
20
21
22def transform_playbacks(
23    df: pd.DataFrame,
24    content_durations: dict | None = None,  # {content_id: duracion_en_SEGUNDOS}
25) -> pd.DataFrame:
26    """Transforma playbacks raw en formato silver limpio."""
27    if df.empty:
28        return df
29
30    initial_count = len(df)
31
32    # 1. Parsear timestamps con timezone. VA PRIMERO: la deduplicacion de abajo
33    #    necesita el instante ya como fecha, no como texto. Y created_at tambien
34    #    se convierte, aunque no se use en los calculos: de esa columna sale la
35    #    marca de agua al final del pipeline, y si llega como texto (que es lo que
36    #    pasa al leer de un CSV en local) el .max().isoformat() del paso 4 revienta
37    #    DESPUES de haber escrito ya en S3.
38    df['started_at'] = pd.to_datetime(df['started_at'], utc=True)
39    df['ended_at'] = pd.to_datetime(df['ended_at'], utc=True, errors='coerce')
40    df['created_at'] = pd.to_datetime(df['created_at'], utc=True)
41
42    # 1 bis. Sin usuario, sin contenido o sin instante de inicio no se puede
43    #        atribuir la reproduccion a nada. Y sin instante, ademas, event_date
44    #        sale a nulo y la carga escribe una particion llamada "event_date=NaT":
45    #        basura que aparece en el lake seis meses despues y nadie sabe de donde
46    #        salio.
47    antes = len(df)
48    df = df[df['user_id'].notna() & df['content_id'].notna() & df['started_at'].notna()]
49    if len(df) < antes:
50        logger.warning(f"Descartadas {antes - len(df)} filas sin usuario, contenido o inicio")
51
52    # 2. Deduplicar.
53    #
54    #    OJO con la clave: deduplicar por el identificador de la fila no quita
55    #    nada, porque por definicion es unico. Los duplicados que este caso tiene
56    #    (~4% en iOS) son el MISMO evento mandado dos veces por el reproductor, y
57    #    llegan con identificadores distintos. Lo que los define es la sesion, el
58    #    tipo de evento y el instante -- y como el reintento llega con unos
59    #    segundos de diferencia, hay que redondear el instante antes de agrupar.
60    #
61    #    El margen de 10 s hay que ajustarlo mirando el fichero: si los duplicados
62    #    son exactos, sobra; si el reintento llega un minuto despues, se queda corto.
63    #
64    #    Y que columnas identifican "el mismo evento" depende de la fuente:
65    #      - En el fichero de eventos del caso: la sesion y el tipo de evento.
66    #      - En la tabla playbacks del esquema de Daniel, que no tiene ninguna de
67    #        las dos: usuario y contenido. Es una clave peor -- un usuario que ve
68    #        dos veces el mismo titulo en el mismo intervalo de 10 s cuenta como
69    #        uno -- pero a 10 s de margen eso no pasa en la vida real, y es lo
70    #        que se puede hacer con lo que hay.
71    #    Se decide mirando las columnas, no suponiendo.
72    if len(df):
73        antes = len(df)
74        claves = [c for c in ('playback_session_id', 'event_type') if c in df.columns]
75        if not claves:
76            claves = ['user_id', 'content_id']
77            logger.warning(
78                f"Sin sesion ni tipo de evento: deduplicando por {claves} + instante"
79            )
80        df['_dedup'] = df['started_at'].dt.floor('10s').astype(str)
81        for col in claves:
82            df['_dedup'] = df['_dedup'] + '|' + df[col].astype(str)
83        df = df.drop_duplicates(subset=['_dedup'], keep='first').drop(columns=['_dedup'])
84        if len(df) < antes:
85            logger.info(f"Duplicados eliminados: {antes - len(df)}")
86
87    # 3. Validar watch_seconds (no negativo, no mayor que 24h)
88    df['watch_seconds'] = pd.to_numeric(
89        df['watch_seconds'], errors='coerce'
90    )
91    # OJO: aqui NO se hace .fillna(0).
92    #
93    # Convertir "no se" en "cero" pierde la diferencia, y aqui cuesta dinero:
94    # una fila con watch_seconds nulo saldria como reproduccion de CERO segundos,
95    # pasaria el filtro de abajo (cero es >= cero) y entraria en el denominador
96    # de la tasa de finalizacion y del tiempo medio por sesion -- dos de las cinco
97    # metricas que van a la reunion del viernes. Un punado de nulos convertidos en
98    # ceros baja la media sin que nadie sepa por que.
99    #
100    # Un nulo se rellena cuando sabes con que. Si no lo sabes, se queda nulo y
101    # quien calcule la media decide que hacer con el.
102    # Fuera lo imposible, pero SIN arrastrar lo desconocido: los nulos se quedan.
103    df = df[df['watch_seconds'].isna() | df['watch_seconds'].between(0, 86400)]
104    # Entero que admite nulos: si se dejara en float, el tipo cambia en el Parquet.
105    df['watch_seconds'] = df['watch_seconds'].astype('Int64')
106    sin_tiempo = int(df['watch_seconds'].isna().sum())
107    if sin_tiempo:
108        logger.warning(f"{sin_tiempo} reproducciones sin tiempo de visionado: quedan a nulo")
109
110    # 4. Calcular campos derivados
111    #
112    # Guardamos y particionamos por fecha UTC: asi el lake es estable y no cambia
113    # si algun dia la empresa cambia de sitio. Pero "un dia" para el negocio es un
114    # dia en Madrid, y la diferencia no es teorica: todo lo que pasa entre las 00:00
115    # y la 01:00 peninsular en invierno (y hasta las 02:00 en verano) cae en el dia
116    # ANTERIOR en UTC. El DAU que Marta ensena el viernes sale de aqui.
117    #
118    # Se dejan las dos columnas: una para guardar, otra para contar.
119    df['event_date'] = df['started_at'].dt.date
120    df['event_date_local'] = df['started_at'].dt.tz_convert('Europe/Madrid').dt.date
121    df['event_hour'] = df['started_at'].dt.hour
122    df['day_of_week'] = df['started_at'].dt.dayofweek  # 0=lunes
123    df['is_weekend'] = df['day_of_week'].isin([5, 6])
124
125    # 5. Que fraccion del titulo se vio.
126    #
127    # Sin esta columna, "finalizacion" solo puede salir del flag completed, que
128    # es un si/no puesto por el reproductor. El ratio dice cuanto se vio de
129    # verdad, y es lo que el miercoles permite responder a Sofia si las peliculas
130    # largas se terminan o se abandonan a la mitad.
131    #
132    # Se calcula AQUI y no al consultar porque necesita la duracion del titulo,
133    # que vive en otra tabla: hacerlo una vez en la carga ahorra un JOIN en cada
134    # consulta posterior. Y si no se conoce la duracion, el ratio es NULO --
135    # nunca cero, que significaria "no vio nada".
136    if content_durations:
137        duracion = df['content_id'].map(content_durations)
138    else:
139        duracion = pd.Series(pd.NA, index=df.index, dtype='Float64')
140    df['completion_ratio'] = (df['watch_seconds'] / duracion.where(duracion > 0)).round(3)
141    # Tope en 1.5: rebobinar y volver a ver un trozo da ratios por encima de 1,
142    # y eso es real. Un ratio de 40 no: es una duracion mal puesta en el catalogo.
143    df['completion_ratio'] = df['completion_ratio'].clip(upper=1.5)
144
145    # 6. Categorizar la sesion.
146    #
147    # Estos nombres son los que van a usar las consultas del resto de la semana
148    # (el jueves se filtra session_category = 'bounce' para no inflar el DAU),
149    # asi que se escriben una vez aqui y no se vuelven a inventar.
150    # Si no se sabe el tiempo, la categoria es 'unknown', no 'bounce'.
151    # El .fillna(False) no es decorativo: con watch_seconds nulo, la comparacion
152    # devuelve nulo (no False), y np.select no sabe que hacer con un nulo.
153    condiciones = [
154        (df['watch_seconds'] < 30).fillna(False),
155        df['watch_seconds'].between(30, 300, inclusive='left').fillna(False),
156        df['watch_seconds'].between(300, 1800, inclusive='left').fillna(False),
157        df['watch_seconds'].between(1800, 7200, inclusive='left').fillna(False),
158        (df['watch_seconds'] >= 7200).fillna(False),
159    ]
160    etiquetas = ['bounce', 'short', 'medium', 'long', 'binge']
161    df['session_category'] = np.select(condiciones, etiquetas, default='unknown')
162
163    # 6. Flag de sesión significativa (> 2 min de watch).
164    # Si no sabemos el tiempo, no sabemos si fue significativa: nulo, no "no".
165    # (Con dtype boolean de pandas, el nulo se propaga solo.)
166    # El umbral va en una constante (arriba del fichero) y no repetido en cada
167    # consulta. Manana el dia 3 calcula el DAU filtrando por esta columna, no
168    # escribiendo otra vez un numero: el dia que alguien decida que son 45
169    # segundos se cambia aqui y no en siete consultas de dos lecciones.
170    df['is_meaningful'] = (df['watch_seconds'] > UMBRAL_SESION_SIGNIFICATIVA).astype('boolean')
171    df.loc[df['watch_seconds'].isna(), 'is_meaningful'] = pd.NA
172
173    final_count = len(df)
174    dropped = initial_count - final_count
175    if dropped > 0:
176        logger.warning(f"Descartadas {dropped} filas inválidas ({dropped/initial_count*100:.1f}%)")
177
178    logger.info(f"Transformación completada: {final_count:,} filas válidas")
179    return df.reset_index(drop=True)
180
181
182def transform_users(df: pd.DataFrame) -> pd.DataFrame:
183    """Limpia y enriquece la dimensión de usuarios."""
184    df['registered_at'] = pd.to_datetime(df['registered_at'], utc=True)
185    df['last_login_at'] = pd.to_datetime(df['last_login_at'], utc=True, errors='coerce')
186
187    # NO se guarda la antiguedad en dias.
188    #
189    # Seria (ahora - registered_at).dt.days, y eso depende de CUANDO corre el
190    # pipeline. Guardarlo en el lake lo vuelve irreproducible: reprocesar los
191    # datos de la semana pasada daria valores distintos de los que habia la
192    # semana pasada.
193    #
194    # Regla: en el lake no se guarda nada que dependa del momento de ejecucion.
195    # Se guarda la fecha de alta -- que es un hecho y no cambia -- y la antiguedad
196    # se calcula AL CONSULTAR. Una linea menos aqui y un problema menos en seis meses.
197
198    # Cohorte mensual de registro (esta si: sale de la fecha de alta, no del reloj)
199    df['registration_cohort'] = df['registered_at'].dt.to_period('M').astype(str)
200
201    # Estandarizar plan
202    df['plan'] = df['plan'].fillna('free').str.lower().str.strip()
203
204    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    lote_desde=None,
20    lote_hasta=None,
21):
22    """Escribe un DataFrame como Parquet en S3, opcionalmente particionado.
23
24    lote_desde / lote_hasta: rango del lote que se esta cargando. Se usan para
25    nombrar el fichero cuando hay particion, para que dos ejecuciones no se
26    pisen. Obligatorios si se pasa partition_col.
27    """
28    if df.empty:
29        logger.warning(f"DataFrame vacío, no se escribe en {prefix}")
30        return
31
32    if partition_col and partition_col in df.columns:
33        # Una partición por valor único.
34        #
35        # OJO CON EL NOMBRE DEL FICHERO: aquí es donde se pierden datos.
36        # Si se llama siempre "data.parquet", cada ejecución REESCRIBE entera
37        # la partición de esa fecha con lo que traiga en ese momento. Y como la
38        # marca de agua avanza por created_at (cuándo se insertó la fila) y la
39        # partición se hace por event_date (cuándo empezó la reproducción), que
40        # no son lo mismo, la ejecución de mañana vuelve a tocar la partición de
41        # hoy y borra lo que escribió la de hoy. Resultado: cada noche se pierden
42        # las reproducciones de entre medianoche y las 03:00, para siempre y sin
43        # un solo error en el log.
44        #
45        # El nombre sale del RANGO DEL LOTE, no del reloj: dos ejecuciones
46        # distintas escriben ficheros distintos (nadie pisa a nadie) y relanzar
47        # el MISMO lote reescribe el suyo y sólo el suyo -- que es lo que hace
48        # que la carga se pueda repetir sin pensar.
49        marca = f"{lote_desde:%Y%m%dT%H%M%S}-{lote_hasta:%Y%m%dT%H%M%S}"
50        for value, partition_df in df.groupby(partition_col):
51            key = f"{prefix}/{partition_col}={value}/part-{marca}.parquet"
52            _write_single_parquet(partition_df, s3_client, key)
53    else:
54        # Las dimensiones (usuarios, catálogo) se extraen ENTERAS cada noche, así
55        # que aquí el nombre fijo es lo correcto: cada ejecución sustituye la foto
56        # anterior. Si se les pusiera nombre por lote, el lake acumularía una copia
57        # del padrón por ejecución y los cruces multiplicarían filas. Son dos
58        # escrituras con dos semánticas distintas.
59        key = f"{prefix}/data.parquet"
60        _write_single_parquet(df, s3_client, key)
61
62
63def _write_single_parquet(df: pd.DataFrame, s3_client, key: str):
64    """Escribe un único archivo Parquet a S3."""
65    table = pa.Table.from_pandas(df)
66    buffer = BytesIO()
67    pq.write_table(table, buffer, compression='snappy')
68    buffer.seek(0)
69
70    s3_client.put_object(
71        Bucket=config.s3_bucket,
72        Key=key,
73        Body=buffer.getvalue(),
74        ContentType='application/octet-stream'
75    )
76    size_mb = buffer.getbuffer().nbytes / (1024 * 1024)
77    logger.info(f"Escrito s3://{config.s3_bucket}/{key} ({size_mb:.2f} MB)")
78
79
80def contar_filas_del_lake() -> int:
81    """Cuenta lo que hay en el lake, sin tocar el fichero de Metabase.
82
83    Conexion en memoria y de usar y tirar: lee el Parquet de S3 directamente,
84    asi que no abre ni bloquea nada.
85    """
86    con = duckdb.connect(":memory:")
87    con.execute("INSTALL httpfs; LOAD httpfs;")
88    con.execute(f"SET s3_region = '{config.s3_region}';")
89    total = con.execute(
90        f"SELECT COUNT(*) FROM read_parquet('s3://{config.s3_bucket}/silver/playbacks/**/*.parquet')"
91    ).fetchone()[0]
92    con.close()
93    return total
94
95
96def setup_duckdb_analytics():
97    """Crea las vistas del lake. SE EJECUTA UNA VEZ, a mano, en el montaje.
98
99    No forma parte del pipeline nocturno, y esto es la decision del dia 1: si el
100    script de cada noche abriera este fichero mientras Metabase lo tiene abierto,
101    uno de los dos se queda fuera. Las vistas apuntan al Parquet de S3, asi que
102    los datos nuevos se ven sin tocar el fichero: no hay nada que refrescar.
103    Metabase lo abre en modo lectura.
104    """
105    con = duckdb.connect(config.duckdb_path)
106
107    # Instalar y cargar extensión para S3
108    con.execute("INSTALL httpfs; LOAD httpfs;")
109    con.execute(f"SET s3_region = '{config.s3_region}';")
110
111    # Crear vistas sobre los archivos Parquet en S3
112    bucket = config.s3_bucket
113    con.execute(f"""
114        CREATE OR REPLACE VIEW silver_playbacks AS
115        SELECT * FROM read_parquet(
116            's3://{bucket}/silver/playbacks/event_date=*/*.parquet',
117            hive_partitioning = true
118        );
119    """)
120    con.execute(f"""
121        CREATE OR REPLACE VIEW dim_users AS
122        SELECT * FROM read_parquet('s3://{bucket}/silver/users/data.parquet');
123    """)
124    con.execute(f"""
125        CREATE OR REPLACE VIEW dim_content AS
126        SELECT * FROM read_parquet('s3://{bucket}/silver/content/data.parquet');
127    """)
128
129    # Y las tres que hacen falta manana para hablar de dinero y de crecimiento.
130    # No pasan por la transformacion porque no vienen de la base de datos: los
131    # cobros salen de una exportacion de la pasarela de pago y el gasto de
132    # marketing lo mantiene alguien a mano en una hoja. Se leen tal cual y se
133    # limpian al consultar.
134    #
135    # Ponerlas aqui, y no manana, es la diferencia entre que la leccion de
136    # metricas empiece calculando o empiece buscando de donde sale cada tabla.
137    for nombre, fichero in [
138        ('suscripciones', 'suscripciones.csv'),
139        ('pagos', 'pagos.csv'),
140        ('marketing', 'gasto_marketing_semanal.csv'),
141    ]:
142        con.execute(f"""
143            CREATE OR REPLACE VIEW {nombre} AS
144            SELECT * FROM read_csv_auto('s3://{bucket}/raw/{fichero}');
145        """)
146
147    # Verificar que funciona
148    count = con.execute("SELECT COUNT(*) FROM silver_playbacks").fetchone()[0]
149    logger.info(f"DuckDB configurado. silver_playbacks: {count:,} filas")
150
151    con.close()
152    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
6
7import pandas as pd   # hace falta para pd.Timestamp() de mas abajo
8from .config import config
9from .extract import extract_playbacks, extract_users, extract_content, get_watermark
10from .transform import transform_playbacks, transform_users
11from .load import write_parquet_to_s3, contar_filas_del_lake
12
13logger = logging.getLogger(__name__)
14
15
16def run_daily_pipeline():
17    """Pipeline diario completo: extract -> transform -> load."""
18    start_time = datetime.now(timezone.utc)
19    logger.info(f"=== Pipeline iniciado: {start_time.isoformat()} ===")
20
21    s3 = boto3.client('s3', region_name=config.s3_region)
22
23    try:
24        # 1. EXTRACT
25        watermark = get_watermark(s3, 'playbacks')
26        logger.info(f"Watermark playbacks: {watermark.isoformat()}")
27
28        playbacks_raw = extract_playbacks(since=watermark)
29        users_raw = extract_users()
30        content_raw = extract_content()
31
32        if playbacks_raw.empty:
33            logger.info("No hay nuevos playbacks. Pipeline finalizado.")
34            return {"status": "no_new_data", "duration_seconds": 0}
35
36        # 2. TRANSFORM
37        #    Las duraciones del catalogo entran aqui para poder calcular que
38        #    fraccion de cada titulo se vio. OJO CON LA UNIDAD: el ratio divide
39        #    segundos entre segundos. Si la columna del origen viniera en minutos
40        #    (es lo que pasa con el fichero del caso), hay que multiplicar por 60
41        #    ANTES de construir el diccionario, no despues de ver los numeros.
42        duraciones = dict(
43            zip(content_raw['content_id'], content_raw['duration_seconds'])
44        )
45        playbacks_clean = transform_playbacks(playbacks_raw, content_durations=duraciones)
46        users_clean = transform_users(users_raw)
47
48        # 3. LOAD a S3.
49        #    Al escribir los playbacks se pasa el RANGO del lote: de ahi sale el
50        #    nombre del fichero, para que dos ejecuciones no se pisen y relanzar
51        #    el mismo lote sea inofensivo.
52        write_parquet_to_s3(
53            playbacks_clean, s3,
54            prefix=f"{config.silver_prefix}/playbacks",
55            partition_col='event_date',
56            lote_desde=pd.Timestamp(watermark),
57            lote_hasta=playbacks_clean['created_at'].max(),
58        )
59        # Las dimensiones van sin rango: se reescriben enteras cada noche.
60        write_parquet_to_s3(users_clean, s3, prefix=f"{config.silver_prefix}/users")
61        write_parquet_to_s3(content_raw, s3, prefix=f"{config.silver_prefix}/content")
62
63        # 4. Actualizar watermark
64        new_watermark = playbacks_clean['created_at'].max().isoformat()
65        _update_watermark(s3, 'playbacks', new_watermark)
66
67        # 5. Comprobar que el lake se lee. NO se toca el fichero .duckdb de
68        #    Metabase: las vistas apuntan al Parquet de S3, asi que los datos
69        #    nuevos aparecen solos y no hay nada que refrescar. Se abre una
70        #    conexion en memoria, de usar y tirar, para contar lo que hay.
71        total_rows = contar_filas_del_lake()
72
73        duration = (datetime.now(timezone.utc) - start_time).total_seconds()
74        result = {
75            "status": "success",
76            "new_playbacks": len(playbacks_clean),
77            "total_rows_in_lake": total_rows,
78            "duration_seconds": round(duration, 1),
79        }
80        logger.info(f"=== Pipeline completado: {result} ===")
81        return result
82
83    except Exception as e:
84        logger.error(f"Pipeline FALLIDO: {e}", exc_info=True)
85        # NO devolvemos un diccionario de error: lo relanzamos.
86        #
87        # Si aqui se devuelve un dict con status "error", el proceso termina con
88        # codigo de salida CERO. Para cron, para el sistema y para cualquier cosa
89        # que mire si el trabajo fue bien, la ejecucion ha sido un exito. El unico
90        # rastro es una linea en un log que nadie abre hasta que algo se nota --
91        # normalmente el jueves por la tarde, mirando un panel con los datos de
92        # anteayer, una hora antes de preparar la reunion.
93        #
94        # Capturar una excepcion y seguir como si nada no es gestionar un error:
95        # es esconderlo. La primera regla de un pipeline nocturno es que alguien
96        # se entere de que ha fallado ANTES que la persona que usa el dato.
97        raise
98
99
100def _update_watermark(s3_client, table_name: str, timestamp_iso: str):
101    """Actualiza el watermark en S3."""
102    try:
103        obj = s3_client.get_object(Bucket=config.s3_bucket, Key=config.watermark_key)
104        watermarks = json.loads(obj['Body'].read())
105    except s3_client.exceptions.NoSuchKey:
106        watermarks = {}
107
108    watermarks[table_name] = timestamp_iso
109    s3_client.put_object(
110        Bucket=config.s3_bucket,
111        Key=config.watermark_key,
112        Body=json.dumps(watermarks, indent=2).encode(),
113    )

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

1# cron/crontab.txt
2#
3# UN SOLO trabajo, no tres. Las metricas van DETRAS del pipeline en la misma
4# orden, no media hora despues y cruzando los dedos: si una noche el pipeline
5# tarda mas de lo normal (red lenta, lote grande, la replica ocupada con el
6# backup), unas metricas lanzadas por reloj se calculan sobre un lake a medias
7# y nadie se entera.
8#
9# Y si algo falla, un correo. No hace falta monitorizacion sofisticada en una
10# empresa de 30 personas: hace falta que alguien se entere antes que Marta.
110 3 * * * cd /opt/streamvibe-data && venv/bin/python -m etl.daily >> /var/log/streamvibe-etl.log 2>&1 || echo "El ETL de StreamVibe ha fallado, mira /var/log/streamvibe-etl.log" | mail -s "[ETL] fallo" tu@streamvibe.es
12
13# Copia de seguridad: NO se copia el fichero .duckdb.
14#
15# Ese fichero solo contiene tres VISTAS sobre el Parquet de S3 -- unos kilobytes
16# de definiciones, cero datos. Copiarlo cada noche no protege nada. Los datos
17# estan en S3, y ahi lo que protege es el versionado del bucket, que guarda las
18# versiones anteriores de un fichero sobrescrito o borrado por error:
19#
20#   aws s3api put-bucket-versioning --bucket streamvibe-data-lake \
21#     --versioning-configuration Status=Enabled
22#
23# Ojo: el versionado guarda TODAS las versiones antiguas y se pagan, asi que va
24# siempre con una regla de ciclo de vida que las expire a los 30 dias.

Un solo job: el pipeline y, si ha ido bien, las métricas. Con aviso por correo si falla.

1# etl/daily.py -- lo que lanza cron: un lote y, si ha ido bien, las metricas
2from .pipeline import run_daily_pipeline
3from ..analytics.metrics import recalcular_metricas
4
5resultado = run_daily_pipeline()
6if resultado["status"] == "success":
7    recalcular_metricas()
8
9
10# etl/backfill.py -- la carga historica: se lanza A MANO, hoy, una sola vez
11def backfill(max_lotes: int = 50):
12    """Carga el historico entero, lote a lote, sin parar hasta terminar."""
13    for n in range(1, max_lotes + 1):
14        resultado = run_daily_pipeline()
15        if resultado["status"] == "no_new_data":
16            logger.info(f"Historico completo tras {n - 1} lotes.")
17            recalcular_metricas()   # una sola vez, al final
18            return
19        logger.info(f"Lote {n} cargado ({resultado['rows_loaded']:,} filas). Sigo.")
20    raise RuntimeError(
21        f"Sigue habiendo datos tras {max_lotes} lotes: revisa la marca de agua"
22    )
23
24# El tope no es adorno: si el pipeline fallara devolviendo un error en vez de
25# levantarlo, un bucle sin tope reintentaria para siempre contra la replica que
26# Daniel pidio no saturar. Con la excepcion relanzada, el bucle se corta solo;
27# el tope es el cinturon por si acaso.

Dos puntos de entrada distintos: el diario que lanza cron, y el histórico que lanzas tú hoy.

1# tests/test_transform.py
2import pandas as pd
3import pytest
4from etl.transform import transform_playbacks
5
6
7# Helper: una fila completa. Si un test omite una columna que la transformacion
8# toca, el test revienta con KeyError antes de llegar al assert -- y entonces no
9# esta probando lo que crees que prueba, solo que no explota.
10def fila(**kwargs):
11    base = {
12        'playback_session_id': 'sess_1',
13        'event_type': 'play',
14        'user_id': 10,
15        'content_id': 100,
16        'started_at': '2024-03-15T20:00:00Z',
17        'ended_at': '2024-03-15T20:30:00Z',
18        'watch_seconds': 1000,
19        'completed': False,
20        'created_at': '2024-03-15T21:00:00Z',
21    }
22    base.update(kwargs)
23    return base
24
25
26def test_deduplication():
27    """El mismo evento mandado dos veces (reintento de iOS) se queda en uno."""
28    df = pd.DataFrame([
29        fila(playback_session_id='s1'),
30        fila(playback_session_id='s1'),                      # duplicado
31        fila(playback_session_id='s2', user_id=20),
32        fila(playback_session_id='s3', user_id=30),
33    ])
34    result = transform_playbacks(df)
35    assert len(result) == 3
36
37
38def test_invalid_watch_seconds_filtered():
39    """watch_seconds negativos o > 24h se descartan."""
40    df = pd.DataFrame([
41        fila(playback_session_id='s1', watch_seconds=-5),      # imposible
42        fila(playback_session_id='s2', watch_seconds=90000),   # > 24h
43        fila(playback_session_id='s3', watch_seconds=1000),    # valido
44    ])
45    result = transform_playbacks(df)
46    assert len(result) == 1
47    assert result.iloc[0]['watch_seconds'] == 1000
48
49
50def test_watch_seconds_desconocido_no_es_cero():
51    """Un tiempo desconocido se queda a NULO, no se convierte en cero.
52
53    Es el test que protege las metricas del viernes: si el nulo pasara a cero,
54    esa fila entraria en el denominador del tiempo medio y de la tasa de
55    finalizacion, bajando la media sin que nadie sepa por que.
56    """
57    df = pd.DataFrame([
58        fila(playback_session_id='s1', watch_seconds=None),
59        fila(playback_session_id='s2', watch_seconds=600),
60    ])
61    result = transform_playbacks(df)
62    assert len(result) == 2                          # la fila NO se descarta
63    assert result['watch_seconds'].isna().sum() == 1  # sigue siendo desconocida
64    assert pd.isna(result.loc[result['watch_seconds'].isna(), 'is_meaningful'].iloc[0])
65
66
67def test_null_user_filtered():
68    """Registros sin user_id, sin contenido o sin inicio se descartan."""
69    df = pd.DataFrame([
70        fila(playback_session_id='s1', user_id=None),
71        fila(playback_session_id='s2', content_id=None),
72        fila(playback_session_id='s3', started_at=None),
73        fila(playback_session_id='s4', user_id=20),
74    ])
75    result = transform_playbacks(df)
76    assert len(result) == 1
77    assert result.iloc[0]['user_id'] == 20
78
79
80def test_una_sola_fila():
81    """Un lote de una fila no debe romper nada (caso borde tipico)."""
82    result = transform_playbacks(pd.DataFrame([fila()]))
83    assert len(result) == 1
84
85
86def test_empty_dataframe():
87    """Un DataFrame vacío retorna vacío sin error."""
88    df = pd.DataFrame()
89    result = transform_playbacks(df)
90    assert len(result) == 0
91
92
93def test_cargar_el_mismo_lote_dos_veces_no_cambia_el_lake(tmp_path):
94    """El test que habria cazado el peor fallo de este pipeline.
95
96    Cargar dos veces el mismo lote tiene que dejar el lake IGUAL. Si el nombre
97    del fichero de salida fuera fijo ("data.parquet"), la segunda carga
98    reescribiria la particion y se perderian las filas que la primera habia
99    dejado ahi. Con el nombre derivado del rango del lote, la segunda pasada
100    reescribe su propio fichero y nada mas.
101    """
102    df = transform_playbacks(pd.DataFrame([
103        fila(playback_session_id='s1'),
104        fila(playback_session_id='s2', user_id=20),
105    ]))
106    desde = pd.Timestamp('2024-03-15T03:00:00Z')
107    hasta = pd.Timestamp('2024-03-16T03:00:00Z')
108
109    escribir_local(df, tmp_path, 'event_date', desde, hasta)
110    filas_primera = contar_filas(tmp_path)
111    escribir_local(df, tmp_path, 'event_date', desde, hasta)   # mismo lote otra vez
112    assert contar_filas(tmp_path) == filas_primera

Tests que sí pasan, y dos nuevos: el del tiempo desconocido y el de la carga repetida.