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

1"""
2Pipeline de Streaming NeoBank -- Producción
3Job: neobank-txn-streaming-v2
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-v2") \
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    .outputMode("update") \
126    .trigger(processingTime=TRIGGER_INTERVAL) \
127    .queryName("hot-features") \
128    .start()
129
130print("=== PIPELINE DE STREAMING INICIADO ===")
131print(f"  Kafka: {KAFKA_TOPIC} @ {KAFKA_BROKERS}")
132print(f"  Output: {OUTPUT_SILVER}")
133print(f"  Trigger: cada {TRIGGER_INTERVAL}")
134print(f"  Watermark: {WATERMARK_DELAY}")
135print(f"  Deduplicación: por transaction_id dentro del watermark")
136print(f"  Queries activas: silver-writer, hot-features")
137
138# Esperar terminación (en producción corre indefinidamente)
139# spark.streams.awaitAnyTermination()

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

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