lección 5

Día 3 (tarde) — Implementar Spark Structured Streaming

Implementas el pipeline completo de streaming: Kafka a S3 silver, watermarks para late data, deduplicación y el sink a Redis para features.

⏱ 45 min

### Jueves, 14:30 — De prototipo a pipeline real

Con el ADR aprobado y el feature store diseñado, llega el momento de escribir el pipeline completo de streaming. No es un prototipo de demostración — es código que va a correr en producción, procesando 4.5 millones de transacciones al día, 24/7, sin perder una sola. Alberto te recuerda: "En banca, exactly-once no es un nice-to-have. Si procesamos una transacción dos veces, el saldo del cliente está mal. Si no la procesamos, el dinero desaparece."

El pipeline tiene 4 responsabilidades: (1) consumir de Kafka con exactly-once semantics, (2) parsear y validar contra el schema, (3) deduplicar transacciones con ID único, y (4) escribir a S3 silver particionado + alimentar el feature store hot.

1"""
2Pipeline de Streaming NeoBank -- Producción
3Job: neobank-txn-streaming
4Cluster: EMR-streaming-001 (4 executors, 15GB RAM cada uno)
5Input: Kafka MSK topic "transactions.completed"
6Output: S3 silver (Parquet) + Redis (features hot)
7
8Garantías:
9- Exactly-once processing (checkpoints en S3)
10- Tolerancia a late data (watermark 5 min)
11- Deduplicación por transaction_id
12- Auto-recovery en caso de fallo (restart desde checkpoint)
13"""
14
15from pyspark.sql import SparkSession
16from pyspark.sql.functions import (
17    col, from_json, to_timestamp, current_timestamp,
18    window, count, avg, approx_count_distinct,
19    expr, lit, date_format, hour
20)
21from pyspark.sql.types import (
22    StructType, StructField, StringType, DoubleType,
23    TimestampType, BooleanType
24)
25
26# === CONFIGURACIÓN ===
27KAFKA_BROKERS = "b-1.neobank-msk.kafka.eu-west-1.amazonaws.com:9092"
28KAFKA_TOPIC = "transactions.completed"
29CHECKPOINT_BASE = "s3://neobank-checkpoints/streaming-v2/"
30OUTPUT_SILVER = "s3://neobank-lake/silver/transactions/"
31TRIGGER_INTERVAL = "30 seconds"
32WATERMARK_DELAY = "5 minutes"
33
34# === SCHEMA ===
35txn_schema = StructType([
36    StructField("transaction_id", StringType(), False),
37    StructField("account_id", StringType(), False),
38    StructField("merchant_id", StringType(), False),
39    StructField("amount", DoubleType(), False),
40    StructField("currency", StringType(), False),
41    StructField("timestamp", StringType(), False),
42    StructField("channel", StringType(), True),  # app, web, pos, atm
43    StructField("country_code", StringType(), True),
44    StructField("status", StringType(), False),
45    StructField("merchant_category", StringType(), True),
46])
47
48# === SPARK SESSION ===
49spark = SparkSession.builder \
50    .appName("neobank-txn-streaming") \
51    .config("spark.sql.streaming.stateStore.providerClass",
52            "org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider") \
53    .config("spark.sql.shuffle.partitions", "16") \
54    .getOrCreate()
55
56spark.sparkContext.setLogLevel("WARN")
57
58# === PASO 1: LEER DE KAFKA ===
59raw_stream = spark.readStream \
60    .format("kafka") \
61    .option("kafka.bootstrap.servers", KAFKA_BROKERS) \
62    .option("subscribe", KAFKA_TOPIC) \
63    .option("startingOffsets", "latest") \
64    .option("maxOffsetsPerTrigger", 50000) \
65    .option("kafka.session.timeout.ms", "30000") \
66    .option("kafka.max.poll.records", "5000") \
67    .load()
68
69# === PASO 2: PARSEAR Y VALIDAR ===
70parsed = raw_stream \
71    .selectExpr(
72        "CAST(key AS STRING) as kafka_key",
73        "CAST(value AS STRING) as json_str",
74        "topic", "partition", "offset",
75        "timestamp as kafka_timestamp"
76    ) \
77    .select(
78        from_json(col("json_str"), txn_schema).alias("txn"),
79        col("kafka_key"),
80        col("kafka_timestamp"),
81        col("partition").alias("kafka_partition"),
82        col("offset").alias("kafka_offset"),
83    ) \
84    .select("txn.*", "kafka_key", "kafka_timestamp", "kafka_partition", "kafka_offset") \
85    .withColumn("event_time", to_timestamp(col("timestamp"))) \
86    .withColumn("processing_time", current_timestamp()) \
87    .filter(col("status") == "completed") \
88    .filter(col("amount") > 0)  # Filtrar transacciones con amount 0 o negativo
89
90# === PASO 3: DEDUPLICACIÓN POR TRANSACTION_ID ===
91# Spark Structured Streaming soporta dropDuplicates con watermark
92deduplicated = parsed \
93    .withWatermark("event_time", WATERMARK_DELAY) \
94    .dropDuplicatesWithinWatermark(["transaction_id"])
95
96# === PASO 4A: ESCRIBIR A S3 SILVER (particionado por fecha y hora) ===
97silver_query = deduplicated \
98    .withColumn("date_partition", date_format(col("event_time"), "yyyy-MM-dd")) \
99    .withColumn("hour_partition", hour(col("event_time"))) \
100    .writeStream \
101    .format("parquet") \
102    .option("path", OUTPUT_SILVER) \
103    .option("checkpointLocation", f"{CHECKPOINT_BASE}silver/") \
104    .partitionBy("date_partition", "hour_partition") \
105    .trigger(processingTime=TRIGGER_INTERVAL) \
106    .outputMode("append") \
107    .queryName("silver-writer") \
108    .start()
109
110# === PASO 4B: CALCULAR FEATURES HOT ===
111hot_features = deduplicated \
112    .groupBy(
113        col("account_id"),
114        window(col("event_time"), "1 hour", "1 minute")
115    ) \
116    .agg(
117        count("*").alias("txn_count_1h"),
118        avg("amount").alias("avg_amount_1h"),
119        approx_count_distinct("merchant_id").alias("distinct_merchants_1h"),
120    )
121
122# Escribir features a consola (en producción: forEach a Redis)
123features_query = hot_features.writeStream \
124    .format("console") \
125    .option("checkpointLocation", f"{CHECKPOINT_BASE}features/") \
126    .outputMode("update") \
127    .trigger(processingTime=TRIGGER_INTERVAL) \
128    .queryName("hot-features") \
129    .start()
130
131# Cada query necesita SU checkpoint, en su propio directorio. Dos queries
132# compartiendo checkpoint se pisan los offsets y pierden datos en silencio.
133
134print("=== PIPELINE DE STREAMING INICIADO ===")
135print(f"  Kafka: {KAFKA_TOPIC} @ {KAFKA_BROKERS}")
136print(f"  Output: {OUTPUT_SILVER}")
137print(f"  Trigger: cada {TRIGGER_INTERVAL}")
138print(f"  Watermark: {WATERMARK_DELAY}")
139print(f"  Deduplicación: por transaction_id dentro del watermark")
140print(f"  Queries activas: silver-writer, hot-features")
141
142# Esta linea bloquea hasta que alguna query termine o falle. Es lo que
143# mantiene vivo el job: sin ella, el proceso llega al final del fichero,
144# se cierra, y se lleva por delante los dos streams.
145spark.streams.awaitAnyTermination()

Pipeline completo de streaming: Kafka -> parse -> dedup -> S3 silver + features hot

### Watermarks — tolerancia a late data

En un sistema distribuido, los eventos no siempre llegan en orden. Un cliente puede hacer una compra a las 14:00 pero la confirmación del banco del merchant llega a las 14:03 (delay de red, retry del POS terminal, etc.). El watermark le dice a Spark: "acepta eventos hasta 5 minutos de retraso respecto al evento más reciente que has visto. Después de eso, descártalos." Es el trade-off entre completitud y latencia.

El watermark de 5 minutos es una decisión de negocio, no técnica. Alberto y Miguel acordaron que perder el 0.01% de eventos con más de 5 min de retraso es aceptable vs. esperar indefinidamente. Si el regulador exigiera 100% completitud, necesitaríamos un approach diferente (reconciliación batch posterior). Documenta siempre POR QUÉ elegiste un watermark específico.

### Deduplicación en streaming

Kafka garantiza at-least-once delivery por defecto. Eso significa que un mismo evento puede llegar más de una vez (por ejemplo, si el producer hace retry tras un timeout). En banca, procesar una transacción dos veces es inaceptable — el cliente vería un cargo doble. La deduplicación por transaction_id dentro del watermark window resuelve esto:

1# Deduplicación en Spark Structured Streaming
2# dropDuplicatesWithinWatermark mantiene estado solo dentro de la ventana del watermark
3# Esto limita el uso de memoria (vs. dropDuplicates global que crece sin límite)
4
5# SIN deduplicación (peligroso en producción):
6# Si el mismo TXN-12345 llega 2 veces, se procesa 2 veces
7# -> Saldo del cliente incorrecto
8# -> Feature txn_count_1h inflada artificialmente
9# -> Modelo de fraude puede generar false positive
10
11# CON deduplicación dentro de watermark:
12deduplicated = parsed \
13    .withWatermark("event_time", "5 minutes") \
14    .dropDuplicatesWithinWatermark(["transaction_id"])
15
16# Spark mantiene los transaction_ids vistos en los últimos 5 minutos
17# Si ve un duplicado dentro de esa ventana, lo descarta
18# Después de 5 min, el estado se limpia automáticamente
19
20# ¿Por qué "within watermark" y no global?
21# - Global: mantendría TODOS los IDs en memoria -> OOM con 4.5M txns/día
22# - Within watermark: solo mantiene ~15K IDs (5 min * 52 evt/s = 15,600)
23# - Memoria: ~15K * 36 bytes (UUID) = ~560 KB (trivial)
24
25print("Deduplicación configurada:")
26print(f"  Ventana de dedup: {5} minutos (alineada con watermark)")
27print(f"  IDs en memoria (estimado): ~{5*60*52:,} ({5*60*52*36/1024:.0f} KB)")
28print(f"  Duplicados esperados: <0.1% del tráfico total")

Deduplicación dentro del watermark — exactly-once con memoria acotada

dropDuplicates (sin watermark) en Spark Streaming mantiene TODOS los IDs en estado — crece indefinidamente hasta OOM. SIEMPRE usa dropDuplicatesWithinWatermark en producción. La diferencia entre un prototipo que funciona 1 hora y un sistema que corre 6 meses sin caerse está en estos detalles.

Y el aviso simétrico: dropDuplicatesWithinWatermark NO es exactly-once frente a reintentos tardíos. Acota la memoria a cambio de una ventana de protección finita. Si el sistema que te envía los eventos reintenta más tarde que tu watermark, necesitas además una clave única en el destino.

1Alberto (revisando tu PR, 17:40):
2"Bien el dropDuplicatesWithinWatermark. Pero quiero que digas
3en voz alta lo que este código NO cubre, porque mañana te lo
4va a preguntar alguien.
5
6Tu watermark son 5 minutos. Eso significa que si el mismo
7transaction_id vuelve a los 6 minutos, tu dedup ya lo ha
8olvidado y lo deja pasar. Y los reintentos de los sistemas
9de pagos no son de 5 minutos: el POS reintenta a los 15, el
10adquirente a la hora, y una reconciliación nocturna puede
11reenviarte el día entero.
12
13Así que ten claro que tienes DOS niveles:
14
15  - Dentro del watermark: Spark. Barato, en memoria, automático.
16  - Fuera del watermark: la tabla de destino. Un MERGE contra
17    transaction_id, o un índice único que reviente. Más caro,
18    pero es el único que aguanta un reintento de 15 minutos.
19
20Hoy solo tienes el primero. Apúntalo en el ADR como riesgo
21conocido y seguimos."

Alberto señala el hueco del dedup: los reintentos tardíos

Regístrate para guardar tu progreso.