lección 4
Día 3 -- Deduplicación GPS con watermarks
Los datos GPS llegan duplicados porque la app reintenta envíos. Implementas deduplicación en streaming con watermarks de Spark para limpiar el stream sin consumir memoria infinita.
⏱ 45 min
### Miércoles, 8:45 AM -- Carlos te avisa de un problema
Llegas a la oficina y Carlos ya te ha escrito por Slack a las 7:50: "Ey, tengo que contarte algo sobre los eventos GPS que no te dije ayer porque no quería abrumarte el primer día. La app de los repartidores tiene un mecanismo de retry: si envía un punto GPS y no recibe ACK del servidor en 2 segundos, reintenta hasta 3 veces. Eso significa que puedes recibir el MISMO punto GPS 2 o 3 veces. El delivery_id + driver_id + timestamp es la clave única pero es tu responsabilidad deduplicar downstream."
Esto explica algo raro que viste ayer en los logs: algunos repartidores aparecían con 2-3 ETAs calculados en el mismo segundo, con los mismos valores. No afectaba al resultado (el ETA era el mismo) pero era ineficiente -- procesar 3 veces el mismo cálculo y hacer 3 escrituras a DynamoDB con el mismo valor. Con 200 ev/s y ~5% de duplicados, son 10 escrituras/seg innecesarias. En Black Friday con 600 ev/s serían 30 escrituras/seg de basura.
Ahora vamos a usar el dataset eventos_gps_sucios.json (descárgalo del apartado de datasets del caso). Es un volcado del topic driver.location con los problemas reales que tendrás que limpiar: ~4.000 duplicados exactos, ~1.650 eventos con timestamp desordenado, ~275 con GPS drift (coordenadas 0,0 o accuracy > 200m) y ~5.000 que llegan en un segundo esquema del proveedor PartnerFleet (campos renombrados: latitude/longitude en vez de lat/lon, speed_ms en vez de speed_kmh, delivery_id en vez de shipment_id, course_degrees y precision, con un marcador "_schema": "v2"). El archivo eventos_gps.json es la versión ya limpia, por si quieres comparar. Tu pipeline debe normalizar ambos esquemas antes de deduplicar.
### El problema: deduplicar en streaming sin memoria infinita
La deduplicación en batch es trivial: cargas todos los datos, haces un GROUP BY o DISTINCT y listo. En streaming es más complejo porque los datos nunca "terminan" -- no puedes mantener un set infinito de "eventos ya vistos" en memoria. Necesitas una ventana temporal: "recordar los últimos N segundos y descartar duplicados dentro de esa ventana".
Spark Structured Streaming resuelve esto con watermarks. Un watermark es un límite temporal que dice: "solo necesito recordar eventos de los últimos N segundos. Todo lo anterior puede borrarse de la memoria." Combinado con dropDuplicates, obtienes deduplicación eficiente con estado acotado.
### Watermarks: cómo funcionan internamente
Un watermark en Spark Streaming funciona así: Spark rastrea el timestamp máximo visto hasta ahora en el stream. El watermark es ese máximo MENOS un umbral configurable. Todos los eventos con timestamp anterior al watermark se consideran "demasiado tarde" y se descartan. Para deduplicación, Spark solo necesita mantener en memoria los IDs de eventos con timestamp posterior al watermark.
1from pyspark.sql.functions import col, to_timestamp, from_json2from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType34# --- Paso 1: Parsear con event_time explicitamente ---5gps_schema = StructType([6 StructField("driver_id", StringType(), False),7 StructField("lat", DoubleType(), False),8 StructField("lon", DoubleType(), False),9 StructField("timestamp", StringType(), False),10 StructField("speed_kmh", DoubleType(), False),11 StructField("heading", IntegerType(), True),12 StructField("battery_pct", IntegerType(), True),13 StructField("delivery_id", StringType(), True),14])1516parsed_gps = gps_stream \17 .selectExpr("CAST(value AS STRING) as json_str") \18 .select(from_json(col("json_str"), gps_schema).alias("gps")) \19 .select("gps.*") \20 .withColumn("event_time", to_timestamp(col("timestamp")))2122# --- Paso 2: Definir watermark de 30 segundos ---23# "Acepto eventos que lleguen hasta 30 segundos después de su event_time"24# El estado de Spark para deduplicación solo mantiene los últimos 30s25gps_with_watermark = parsed_gps \26 .withWatermark("event_time", "30 seconds")2728# --- Paso 3: Deduplicar por (driver_id, timestamp) ---29# Si dos eventos tienen el mismo driver + mismo timestamp, son duplicados30# dropDuplicates REQUIERE un watermark previo en streaming31gps_deduplicated = gps_with_watermark \32 .dropDuplicates(["driver_id", "timestamp"])3334# Resultado: cada combinacion (driver_id, timestamp) aparece exactamente 1 vez35# Spark purga el estado de dedup para eventos más antiguos que el watermark36# Uso de memoria: ~30 segundos * 200 ev/s * ~500 bytes = ~3 MB (trivial)
Watermark de 30s + dropDuplicates: deduplicación eficiente en streaming
dropDuplicates en streaming REQUIERE un watermark definido previamente. Sin watermark, Spark tendría que mantener TODOS los eventos de toda la historia para detectar duplicados -- memoria infinita. El watermark limita el estado a una ventana temporal acotada. Si olvidas el watermark, Spark te dará un error o consumirá memoria hasta que el executor se cae por OOM.
### Filtros de sanidad: datos GPS imposibles
Además de duplicados, Carlos te comenta otro problema: "Hay smartphones viejos que a veces envían coordenadas (0.0, 0.0) cuando pierden señal GPS. Y algún repartidor con el teléfono en el bolsillo reporta velocidades de 200 km/h por un glitch del acelerómetro." Necesitas filtros que descarten datos físicamente imposibles:
1# Filtros de sanidad para datos GPS2# Descartar: coordenadas fuera de Espana, velocidades imposibles, bateria invalida3clean_gps = gps_deduplicated \4 .filter(5 # Coordenadas dentro de Espana continental + Baleares6 (col("lat").between(36.0, 43.8)) &7 (col("lon").between(-9.3, 4.3)) &8 # Velocidad razonable para un repartidor urbano (moto/bici/furgoneta)9 (col("speed_kmh").between(0, 120)) &10 # Bateria válida11 (col("battery_pct").between(0, 100)) &12 # Solo repartidores con entrega activa13 (col("delivery_id").isNotNull()) &14 # Precision GPS razonable (< 50 metros)15 (col("accuracy_m") < 50)16 )1718# Estadistica: ~3-5% de eventos se descartan por estos filtros19# Es aceptable perder esos datos porque son basura que corromperia los ETAs
Filtros de sanidad: descartar datos GPS fisicamente imposibles
Los filtros de sanidad van DESPUES de la deduplicación y ANTES del cálculo de ETA. El orden importa: primero deduplicar (reducir volumen), luego filtrar (quitar basura), luego calcular (logica de negocio). Si filtras antes de deduplicar, puedes aceptar un duplicado que pasa el filtro pero cuyo original fue descartado.
### El pipeline completo: clean GPS pipeline
Integras watermark + dedup + filtros en un pipeline único que produce un stream limpio de GPS. Este stream limpio alimenta tanto el cálculo de ETAs (DynamoDB) como el histórico en S3 (para análisis posterior):
1from pyspark.sql import SparkSession2from pyspark.sql.functions import (3 col, from_json, to_timestamp, current_timestamp, lit4)56spark = SparkSession.builder \7 .appName("flashroute-gps-clean") \8 .config("spark.sql.streaming.checkpointLocation",9 "s3://flashroute-checkpoints/gps-clean/") \10 .getOrCreate()1112# Ingesta cruda de Kafka13raw_gps = spark.readStream \14 .format("kafka") \15 .option("kafka.bootstrap.servers", "kafka-1.flashroute.internal:9092") \16 .option("subscribe", "driver.location") \17 .option("startingOffsets", "latest") \18 .load()1920# Pipeline completo: parse -> watermark -> dedup -> filter21clean_gps_stream = raw_gps \22 .selectExpr("CAST(value AS STRING) as json_str") \23 .select(from_json(col("json_str"), gps_schema).alias("gps")) \24 .select("gps.*") \25 .withColumn("event_time", to_timestamp(col("timestamp"))) \26 .withWatermark("event_time", "30 seconds") \27 .dropDuplicates(["driver_id", "timestamp"]) \28 .filter(29 (col("lat").between(36.0, 43.8)) &30 (col("lon").between(-9.3, 4.3)) &31 (col("speed_kmh").between(0, 120)) &32 (col("delivery_id").isNotNull())33 )3435# Sink 1: ETA pipeline (consume este stream limpio)36# Sink 2: S3 para histórico (Parquet particionado por fecha)37clean_gps_stream.writeStream \38 .format("parquet") \39 .option("path", "s3://flashroute-lake/silver/gps_clean/") \40 .option("checkpointLocation", "s3://flashroute-checkpoints/gps-s3/") \41 .partitionBy("delivery_date") \42 .trigger(processingTime="60 seconds") \43 .start()
Pipeline completo de limpieza GPS: watermark + dedup + filtros + doble sink
Le enseñas el pipeline a Carlos en un café a las 12:00. Carlos asiente: "Perfecto, esto cubre los reintentos de la app. Hay otro problema que te cuento esta tarde: los GPS que llegan desordenados por túneles. Eso es distinto de los duplicados -- son eventos únicos pero que llegan tarde." Lo dejas para la lección de la tarde.
## ejercicios
Deduplicacion de eventos GPS
Implementa funciones de deduplicación para eventos GPS: primero en Python puro (para entender la logica) y luego indica como se haria en PySpark Streaming con watermarks. Filtra también eventos con datos invalidos.
Analizar latencia optima de watermark
Analiza la distribución de retrasos GPS (diferencia entre event_time y processing_time) para determinar el watermark optimo. Objetivo: capturar >= 99% de eventos sin excesiva latencia.
Regístrate para guardar tu progreso.
## comentarios
Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.
Inicia sesión para comentar y responder.
cargando comentarios...