Día 2 - Pipeline de ingesta y la sorpresa de los duplicados

1#data-team | Miercoles 9:20 AM
2
3Pablo Herrera  9:20 AM
4?? Standup miercoles
5- Marcos: pipeline billing EN REVIEW (PR abierta)
6- Sandra: esperando eventos en Redshift para arrancar modelos
7- Nuevo: pipeline de ingesta de eventos (objetivo: tener datos
8  hoy para que Sandra pueda empezar mañana)
9
10@nuevo como vas? Necesitas algo?
11
12Tu  9:22 AM
13Todo bien! Ayer termine la exploracion y el DDL.
14Hoy objetivo: pipeline funcional que cargue 1 semana de eventos.
15Tengo una duda sobre la estrategia de dedup que me gustaria
16resolver con Marcos o David.
17
18Marcos Delgado  9:24 AM
19Te hago una call a las 10. 5 min.
20
21Sandra Vidal  9:25 AM
22Avisa cuando tengas datos en la tabla! Necesito al menos 4
23semanas de histórico para las cohortes ??
24
25    ?? 2

1# ingest_events.py - Pipeline de ingesta de eventos TaskFlow
2# Lee de S3 (local para desarrollo), limpia y carga a staging
3import pandas as pd
4import json
5from pathlib import Path
6from datetime import datetime, timedelta
7
8# --- Configuracion ---
9RAW_PATH = Path('datos/taskflow_events/')  # Simula S3 en local
10OUTPUT_PATH = Path('output/staging/')
11OUTPUT_PATH.mkdir(parents=True, exist_ok=True)
12
13# Clasificacion de eventos (definición oficial v1)
14CORE_EVENTS = {
15    'task_created', 'task_completed', 'task_assigned',
16    'comment_added', 'file_uploaded', 'board_created',
17    'automation_created',
18}
19BROAD_ONLY = {
20    'board_viewed', 'task_viewed', 'search_performed',
21    'filter_applied', 'notification_clicked',
22    'dashboard_viewed', 'report_exported',
23}
24
25def classify_event(event_type: str) -> str:
26    if event_type in CORE_EVENTS:
27        return 'core'
28    elif event_type in BROAD_ONLY:
29        return 'broad'
30    return 'system'
31
32def extract_property(properties: dict, key: str) -> str:
33    """Extrae un campo del dict de propiedades, None si no existe."""
34    if properties and isinstance(properties, dict):
35        return properties.get(key)
36    return None
37
38# --- Lectura ---
39def read_events_for_date(date_str: str) -> pd.DataFrame:
40    """Lee todos los eventos de un día desde la estructura S3."""
41    all_events = []
42    date_path = RAW_PATH / date_str.replace('-', '/')
43    
44    if not date_path.exists():
45        print(f"  [WARN] No hay datos para {date_str}")
46        return pd.DataFrame()
47    
48    for jsonl_file in date_path.glob('*.jsonl'):
49        with open(jsonl_file, 'r') as f:
50            for line in f:
51                try:
52                    event = json.loads(line.strip())
53                    all_events.append(event)
54                except json.JSONDecodeError:
55                    continue  # Lineas malformadas se descartan
56    
57    return pd.DataFrame(all_events)
58
59# --- Transformacion ---
60def transform_events(df: pd.DataFrame, source_file: str) -> pd.DataFrame:
61    """Aplica clasificacion, aplanamiento y metadatos."""
62    if df.empty:
63        return df
64    
65    df['event_category'] = df['event_type'].apply(classify_event)
66    df['event_timestamp'] = pd.to_datetime(df['timestamp'])
67    df['event_date'] = df['event_timestamp'].dt.date
68    df['task_id'] = df['properties'].apply(
69        lambda p: extract_property(p, 'task_id')
70    )
71    df['board_id'] = df['properties'].apply(
72        lambda p: extract_property(p, 'board_id')
73    )
74    df['properties_json'] = df['properties'].apply(
75        lambda p: json.dumps(p) if p else None
76    )
77    df['ingested_at'] = datetime.utcnow()
78    df['source_file'] = source_file
79    
80    # Seleccionar columnas finales (orden del DDL)
81    columns = [
82        'event_id', 'event_type', 'event_category',
83        'event_timestamp', 'event_date',
84        'user_id', 'workspace_id',
85        'task_id', 'board_id', 'properties_json',
86        'ingested_at', 'source_file',
87    ]
88    return df[columns]
89
90# --- Deduplicacion ---
91def deduplicate(df: pd.DataFrame) -> pd.DataFrame:
92    """Elimina duplicados por event_id, quedandose con el primero."""
93    before = len(df)
94    df = df.drop_duplicates(subset=['event_id'], keep='first')
95    after = len(df)
96    if before > after:
97        print(f"  [DEDUP] {before - after:,} duplicados eliminados")
98    return df

Pipeline de ingesta: lectura, transformación, deduplicación

1Marcos (call, 10:00 AM):
2
3"Los duplicados. Si. Kafka usa at-least-once delivery por
4defecto. Significa que si el producer no recibe ACK del
5broker (por timeout de red, por ejemplo), reenvia el
6mensaje. El consumer lo recibe dos veces.
7
8Opciones para deduplicar:
91. En el pipeline de ingesta (lo que estas haciendo)
10   - DROP_DUPLICATES por event_id antes de cargar
11   - Simple, funciona, pero si ejecutas dos veces el
12     mismo batch puedes tener problemas
13
142. En Redshift con MERGE/UPSERT
15   - Cargas todo en una staging temporal
16   - Haces MERGE contra la tabla final por event_id
17   - Mas robusto pero más lento
18
193. En dbt (dejar que Sandra lo haga downstream)
20   - La tabla staging tiene duplicados
21   - Sandra deduplica en el primer modelo
22   - Mas simple para ti pero Sandra me mataria
23
24Mi recomendación: opcion 1 + un check de idempotencia.
25Tu pipeline debe poder ejecutarse 2 veces sobre el mismo
26día sin crear duplicados. Usa event_id como clave natural.
27
28Pregunta a David cuantos duplicados esperar. El sabe el
29config exacto del producer."
30
31Tu: "Entendido. Opcion 1 con idempotencia. Le preguntó a
32David el ratio esperado."
33
34Marcos: "Bien. Cuando tengas el PR avisame, lo reviso."

1Tu  10:32 AM (DM a David Martin)
2Hola David! Estoy montando el pipeline de ingesta de eventos
3y he visto duplicados en S3 (~0.25% de los eventos).
4Marcos me dice que es el retry del producer. Sabes cual es
5el config exacto? Quiero saber que ratio de duplicados
6esperar para mi validación.
7
8David Martin  10:48 AM
9Si. El producer está configurado con:
10- acks=all
11- retries=3
12- retry.backoff.ms=100
13- enable.idempotence=false
14
15El enable.idempotence=false es el problema. Con true Kafka
16garantiza exactly-once a nivel de producer pero tiene overhead.
17Lo deje en false cuando lo monte rápido.
18
19El ratio esperado es ~0.2-0.5% segun la carga de red. En
20picos puede subir al 1%.
21
22Quieres que active idempotence? Puedo hacerlo pero necesito
23coordinarlo con infra.
24
25Tu  10:52 AM
26Por ahora no hace falta, deduplico en el pipeline. Pero si
27en el futuro el volumen sube mucho seria bueno tenerlo.
28Gracias por la info!
29
30David Martin  10:53 AM
31np. Si necesitas saber que significa algun evento específico
32preguntame, la documentación del tracking es... escasa.

1# --- Ejecucion principal ---
2def run_ingestion(start_date: str, end_date: str):
3    """Ingesta eventos para un rango de fechas."""
4    print(f"=== Ingesta de eventos: {start_date} a {end_date} ===")
5    
6    start = datetime.strptime(start_date, '%Y-%m-%d')
7    end = datetime.strptime(end_date, '%Y-%m-%d')
8    
9    all_dfs = []
10    current = start
11    while current <= end:
12        date_str = current.strftime('%Y-%m-%d')
13        print(f"\n  Procesando {date_str}...")
14        
15        df = read_events_for_date(date_str)
16        if not df.empty:
17            df = transform_events(df, f"s3://taskflow-events/raw/{date_str}/")
18            df = deduplicate(df)
19            all_dfs.append(df)
20            print(f"  -> {len(df):,} eventos validos")
21        
22        current += timedelta(days=1)
23    
24    if all_dfs:
25        final = pd.concat(all_dfs, ignore_index=True)
26        # Dedup final (por si un evento aparece en 2 archivos de días distintos)
27        final = deduplicate(final)
28        
29        # Guardar como parquet (simula COPY a Redshift)
30        output_file = OUTPUT_PATH / 'stg_product_events.parquet'
31        final.to_parquet(output_file, index=False)
32        
33        print(f"\n=== RESULTADO ===")
34        print(f"Eventos totales: {len(final):,}")
35        print(f"Dias procesados: {(end - start).days + 1}")
36        print(f"Usuarios únicos: {final['user_id'].nunique():,}")
37        print(f"Categorias: {final['event_category'].value_counts().to_dict()}")
38        print(f"Output: {output_file}")
39    else:
40        print("\n[ERROR] No se encontraron eventos en el rango")
41
42# Ejecutar para 1 semana de muestra
43run_ingestion('2024-03-08', '2024-03-14')

Pipeline completo: procesa una semana y genera el staging

1GitHub PR #47 - feat: pipeline de ingesta de eventos de producto
2
3Marcos Delgado (reviewer) - 14:45
4
5src/ingest_events.py:
6L42 - "Has pensado en que pasa si un archivo JSON tiene una linea
7vacia al final? Ese try/except va a tragarsela pero el log no te
8dira nada. Anade un contador de lineas malformadas y logealo al
9final."
10
11L67 - "Bien la dedup por event_id. Una mejora: anade un assert
12después de la dedup que verifique que no hay NULLs en event_id.
13Si Kafka manda un evento sin ID (pasa), tu pipeline deberia
14fallar ruidosamente, no tragar el NULL y crear un duplicado
15potencial."
16
17L89 - "El extract_property está bien pero isinstance(properties,
18dict) va a fallar si el campo es un string JSON en vez de un dict
19parseado. Asegurate de que el json.loads ya se hizo antes."
20
21General: "Buen primer PR. La estructura es clara y el naming sigue
22las convenciones. Los 3 comentarios son mejoras de robustez, no
23bloqueantes. Apruebo con esos cambios. ??"
24
25Sandra Vidal (reviewer) - 15:02
26
27"Los nombres de columnas están perfectos! stg_product_events con
28prefijo de entidad... me haces feliz ??
29Solo una cosa: en el DDL anade un COMMENT a la tabla y a las
30columnas clave. En Redshift se puede con COMMENT ON TABLE.
31Asi cualquiera que haga DESC tabla sabe que es cada campo."

1Marcos Delgado  15:15 PM (DM)
2Mergeado. Buen primer pipeline.
3Una cosa: la proxima vez que ingieras más de 1 semana,
4piensa en particionado. Cargar 6 meses de golpe en un
5solo archivo parquet no va a escalar.
6No es urgente -- es para que lo tengas en mente.
7
8Tu  15:18 PM
9Anotado. Cuando llegue el momento de cargar el histórico
10completo te preguntó el approach.
11
12Marcos Delgado  15:19 PM
13??

1Lucia Fernandez (Meet, 16:05):
2"Ey, tengo dos preguntas para está semana:
3
41. Las cohortes de retención: cuando las tenemos? Necesito
5   incluirlas en el deck para el board review del lunes.
6
72. Pregunta nueva: alguien sabe cuantos usuarios usan la
8   feature de automatizaciones? Es que estamos pensando en
9   moverla al plan Business y necesito saber el impacto."
10
11Pablo: "Las cohortes son el proyecto de está semana. El
12nuevo está montando el pipeline de ingesta -- hoy ha cargado
13los primeros datos. Las cohortes las tenemos el jueves o
14viernes. Sobre las automatizaciones, depende de los eventos."
15
16Tu: "Lucia, la pregunta de automatizaciones la puedo
17responder rápido con los datos que ya tengo cargados.
18Dame hasta mañana por la mañana y te doy un número."
19
20Lucia: "Ey, genial ?? Si puedes segmentarlo por plan
21(free vs pro vs business) mejor."
22
23Sandra: "Ojo con esa segmentacion -- necesitas cruzar
24eventos con la tabla de suscripciones que está en otro
25schema. Hablamos después del sync."
26
27Tu: "Entendido, lo coordino con Sandra."

1Sandra Vidal  17:05 PM (DM)
2Para lo de Lucia -- las suscripciones están en:
3analytics.dim_workspaces
4
5Campos relevantes:
6- workspace_id (PK)
7- workspace_plan: 'free', 'pro', 'business'
8- workspace_created_at
9- workspace_member_count
10- workspace_is_active (bool)
11
12El JOIN es por workspace_id. Cada evento tiene workspace_id
13así que puedes cruzar directamente.
14
15Pero OJO: un workspace puede cambiar de plan. La tabla
16dim_workspaces tiene el plan ACTUAL. Si quieres el plan
17en el momento del evento necesitas la tabla de historial:
18analytics.fct_plan_changes
19
20Para la pregunta de Lucia con el plan actual es suficiente.
21Solo ten cuidado al documentarlo ??
22
23Tu  17:10 PM
24Entendido! Uso dim_workspaces para el plan actual y lo
25documento como "plan a fecha de consulta, no plan en el
26momento del evento". Gracias Sandra!
27
28Sandra Vidal  17:11 PM
29Exacto. Ves, esto es lo que diferencia un número
30"correcto" de un número "correcto Y documentado". ??