lección 2

Día 1 -- Pipeline de ETAs en real-time

Lees eventos GPS de Kafka, calculas distancia restante con la fórmula Haversine e implementas estimación de tiempo de llegada para el dashboard de Ana.

⏱ 45 min

### Lunes, 9:00 AM -- Tu primer standup como owner

El standup de FlashRoute es rápido: 10 minutos de pie junto a la pizarra. Marcos modera. Cada uno dice en una frase qué va a hacer hoy y si tiene bloqueos. No hay "ayer hice X" -- Marcos dice que eso se ve en los commits. Hoy es tu primera semana operativa así que escuchas: Laura habla de una discrepancia en facturación con un cliente de electrónica, Diego de un A/B test en el algoritmo de rutas que mejoró el tiempo medio un 4%, Marta de la integración con un nuevo cliente B2B que empieza el mes que viene, Iván de un problema de stock en el hub de Bilbao.

Cuando termina, Marcos te mira: "Tú hoy: pipeline de ETAs. Ana necesita el dato en DynamoDB para que el frontend lo lea. El servicio de Carlos ya publica GPS en Kafka. Tú consumes, calculas y escribes. Preguntas?" Tienes varias. La más importante: cómo calcular el ETA. Marcos: "Empieza simple. Distancia Haversine dividida por velocidad actual. Factor de corrección urbano para compensar que la ruta no es línea recta. Luego mejora si los números no cuadran."

Es un buen consejo que reconoces de NeoBank: empezar con algo funcional y mejorar iterativamente, no buscar la solución perfecta el día 1. La diferencia es que en NeoBank Alberto te habría dado un diseño detallado. Aquí, Marcos te da el objetivo y tú decides el cómo.

### La fórmula Haversine: distancia entre dos puntos GPS

La fórmula Haversine calcula la distancia en línea recta entre dos puntos sobre la superficie de la Tierra usando sus coordenadas GPS (latitud y longitud). Es la fórmula estándar para navegación y geolocalización cuando las distancias son inferiores a ~500km. Para última milla (distancias de 1-20km) tiene una precisión de +/- 0.5%, más que suficiente.

La fórmula toma como input las coordenadas de dos puntos (en grados) y retorna la distancia en kilómetros. Usa el radio de la Tierra (6371 km) y funciones trigonométricas para tener en cuenta la curvatura terrestre. Para distancias cortas se podría usar pitágoras, pero Haversine es más precisa y el coste computacional es trivial.

1import math
2
3def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
4    """
5    Calcula la distancia en km entre dos puntos GPS usando la formula Haversine.
6    Precision: +/- 0.5% para distancias < 20km (perfecto para última milla).
7    
8    Parametros:
9        lat1, lon1: coordenadas del punto A (grados decimales)
10        lat2, lon2: coordenadas del punto B (grados decimales)
11    Retorna:
12        Distancia en kilometros (linea recta sobre la superficie terrestre)
13    """
14    R = 6371.0  # Radio de la Tierra en km
15
16    # Convertir grados a radianes
17    phi1 = math.radians(lat1)
18    phi2 = math.radians(lat2)
19    delta_phi = math.radians(lat2 - lat1)
20    delta_lambda = math.radians(lon2 - lon1)
21
22    # Formula Haversine
23    a = (math.sin(delta_phi / 2) ** 2 +
24         math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2) ** 2)
25    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
26
27    return R * c
28
29
30# --- Ejemplos con datos reales de FlashRoute ---
31
32# Repartidor DRV-0847: posición actual -> destino del cliente
33dist = haversine_km(40.4168, -3.7038, 40.4015, -3.7105)
34print(f"DRV-0847 a destino: {dist:.3f} km")  # ~1.78 km
35
36# Hub Madrid Centro -> Aeropuerto (referencia larga)
37dist_larga = haversine_km(40.4168, -3.7038, 40.4722, -3.5611)
38print(f"Centro -> Aeropuerto: {dist_larga:.3f} km")  # ~13.2 km
39
40# Factor de correccion urbano: la ruta real es ~1.3-1.5x la linea recta
41# (calles, semaforos, sentidos únicos, rotondas)
42URBAN_FACTOR = 1.4
43dist_real = dist * URBAN_FACTOR
44print(f"Distancia estimada real (x{URBAN_FACTOR}): {dist_real:.3f} km")  # ~2.49 km

Fórmula Haversine -- el cálculo base para distancia entre coordenadas GPS

El factor de corrección urbano de 1.4x es una aproximación razonable para ciudades españolas. En ciudades con grid regular (como Manhattan) sería ~1.2x. En ciudades con calles irregulares (como el casco antiguo de Sevilla) puede llegar a 1.6x. FlashRoute podría calibrar este factor por zona usando datos históricos reales de rutas completadas, pero eso es una optimización futura.

### El pipeline completo: Kafka -> Spark -> DynamoDB

Decides implementar el pipeline con Spark Structured Streaming. Ya tienes experiencia de NeoBank con Spark Streaming y el stack de FlashRoute usa EMR. El flujo es: leer GPS de Kafka, hacer join con destinos de entregas activas, calcular distancia + ETA, escribir a DynamoDB. El trigger es cada 30 segundos para cumplir el SLA de latencia del dashboard de Ana.

1from pyspark.sql import SparkSession
2from pyspark.sql.functions import (
3    col, from_json, udf, current_timestamp, lit, greatest
4)
5from pyspark.sql.types import (
6    StructType, StructField, StringType, DoubleType, IntegerType
7)
8import math
9
10# --- Schema del evento GPS ---
11gps_schema = StructType([
12    StructField("driver_id", StringType(), False),
13    StructField("lat", DoubleType(), False),
14    StructField("lon", DoubleType(), False),
15    StructField("timestamp", StringType(), False),
16    StructField("speed_kmh", DoubleType(), False),
17    StructField("heading", IntegerType(), True),
18    StructField("battery_pct", IntegerType(), True),
19    StructField("accuracy_m", DoubleType(), True),
20    StructField("delivery_id", StringType(), True),
21])
22
23# --- SparkSession con config para streaming ---
24spark = SparkSession.builder \
25    .appName("flashroute-eta-pipeline") \
26    .config("spark.sql.streaming.checkpointLocation",
27            "s3://flashroute-checkpoints/eta-pipeline/") \
28    .config("spark.executor.instances", "4") \
29    .config("spark.executor.cores", "2") \
30    .config("spark.executor.memory", "4g") \
31    .getOrCreate()
32
33# --- UDF de Haversine para usar en DataFrames ---
34@udf(returnType=DoubleType())
35def calc_haversine(lat1, lon1, lat2, lon2):
36    """Calcula distancia en km entre dos puntos GPS."""
37    if any(v is None for v in [lat1, lon1, lat2, lon2]):
38        return None
39    R = 6371.0
40    phi1, phi2 = math.radians(lat1), math.radians(lat2)
41    dp = math.radians(lat2 - lat1)
42    dl = math.radians(lon2 - lon1)
43    a = math.sin(dp/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dl/2)**2
44    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
45
46# --- Leer stream de GPS desde Kafka ---
47gps_stream = spark.readStream \
48    .format("kafka") \
49    .option("kafka.bootstrap.servers", "kafka-1.flashroute.internal:9092") \
50    .option("subscribe", "driver.location") \
51    .option("startingOffsets", "latest") \
52    .option("maxOffsetsPerTrigger", 10000) \
53    .load()
54
55# --- Parsear JSON y filtrar eventos con entrega activa ---
56parsed_gps = gps_stream \
57    .selectExpr("CAST(value AS STRING) as json_str") \
58    .select(from_json(col("json_str"), gps_schema).alias("gps")) \
59    .select("gps.*") \
60    .filter(col("delivery_id").isNotNull())
61
62print(f"Schema parseado: {parsed_gps.schema.simpleString()}")

Primera parte del pipeline: SparkSession, schema, ingesta de Kafka

La segunda parte del pipeline necesita el destino de cada entrega para calcular la distancia restante. En producción, haces un stream-static join con un DataFrame de entregas activas que se refresca periódicamente desde PostgreSQL. Este join es eficiente porque Spark puede broadcastear la tabla de entregas activas (son ~5000 entregas en tránsito en cualquier momento, cabe sobradamente en memoria).

1# --- Lookup de destinos: entregas activas ---
2# Se refresca cada 5 minutos con un job aparte que lee de PostgreSQL
3# En cada micro-batch, el join usa la versión más reciente en cache
4deliveries_df = spark.read \
5    .format("jdbc") \
6    .option("url", "jdbc:postgresql://db.flashroute.internal/ops") \
7    .option("dbtable", "active_deliveries") \
8    .option("user", "etl_reader") \
9    .load() \
10    .select(
11        col("delivery_id"),
12        col("dest_lat"),
13        col("dest_lon"),
14        col("promised_eta"),
15        col("zone"),
16        col("client_b2b")
17    )
18
19# --- Constantes del cálculo de ETA ---
20URBAN_FACTOR = 1.4   # Correccion linea recta -> ruta urbana
21MIN_SPEED_KMH = 5.0  # Velocidad mínima (evita division por 0 cuando para)
22
23# --- Join GPS con destino y calcular ETA ---
24eta_calculated = parsed_gps \
25    .join(deliveries_df, "delivery_id") \
26    .withColumn(
27        "distance_km",
28        calc_haversine(col("lat"), col("lon"), col("dest_lat"), col("dest_lon"))
29        * lit(URBAN_FACTOR)
30    ) \
31    .withColumn(
32        "eta_minutes",
33        (col("distance_km") /
34         greatest(col("speed_kmh"), lit(MIN_SPEED_KMH))) * lit(60)
35    ) \
36    .select(
37        col("delivery_id"),
38        col("driver_id"),
39        col("distance_km"),
40        col("eta_minutes"),
41        col("speed_kmh"),
42        col("zone"),
43        col("client_b2b"),
44        current_timestamp().alias("calculated_at")
45    )

Segunda parte: join con destinos y cálculo de ETA

### El imprevisto: Ana aparece a las 11:30

Estás concentrado escribiendo el sink a DynamoDB cuando Ana Vega aparece detrás de tu pantalla. No es una persona que escriba por Slack cuando puede decírtelo a la cara. "Eres el nuevo de datos, no? Marcos me dijo que estás montando los ETAs. Necesito que entiendas algo: mis coordinadores toman decisiones cada 30 segundos mirando esa pantalla. Si un repartidor va a llegar 15 minutos tarde, necesito saberlo AHORA, no cuando ya haya llegado tarde. Porque si lo sé ahora, puedo reasignar la entrega a otro repartidor que esté más cerca."

Le explicas que el MVP calcula el ETA basado en distancia Haversine y velocidad actual. Ana asiente: "Bien. Pero añade un campo que diga si va a llegar tarde. Comparado con la hora que prometimos al cliente. Eso es lo que mis coordinadores necesitan ver en rojo en la pantalla." Tomas nota mental -- es un requisito claro: campo is_at_risk como booleano que el frontend interpreta como color.

Ana no es técnica pero es extremadamente precisa en lo que necesita. Cuando dice "30 segundos" es literal. Cuando dice "en rojo", quiere un campo booleano que el frontend interprete como color. Aprende su lenguaje: ella piensa en "repartidores retrasados", no en "watermarks" ni "micro-batches". Tu trabajo es traducir sus necesidades de negocio a soluciones técnicas. No la eduques sobre Spark -- dale datos.

### Escribir a DynamoDB -- el sink del pipeline

El último paso es escribir los ETAs calculados a DynamoDB para que el frontend de Ana los lea. DynamoDB es la elección correcta aquí por tres razones: latencia de lectura inferior a 10ms (crítico para un dashboard real-time), escalado automático sin gestión de infra, y el frontend de Carlos ya está integrado con DynamoDB. Usas foreachBatch para escribir en micro-lotes eficientes.

1import boto3
2from datetime import datetime, timedelta
3
4def write_etas_to_dynamodb(batch_df, batch_id):
5    """
6    Escribe los ETAs calculados a DynamoDB.
7    Tabla: delivery_etas (PK: delivery_id)
8    
9    porque permite usar batch_writer de DynamoDB que agrupa
10    hasta 25 items por request HTTP -- mucho más eficiente.
11    """
12    if batch_df.isEmpty():
13        return
14
15    rows = batch_df.collect()
16    dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')
17    table = dynamodb.Table('delivery_etas')
18
19    with table.batch_writer() as writer:
20        for row in rows:
21            writer.put_item(Item={
22                'delivery_id': row['delivery_id'],
23                'driver_id': row['driver_id'],
24                'eta_minutes': int(row['eta_minutes']),
25                'distance_km': str(round(float(row['distance_km']), 2)),
26                'speed_kmh': str(round(float(row['speed_kmh']), 1)),
27                'zone': row['zone'],
28                'is_at_risk': row['is_at_risk'],
29                'calculated_at': row['calculated_at'].isoformat(),
30                'ttl': int((datetime.now() + timedelta(hours=2)).timestamp())
31            })
32
33# --- Iniciar el pipeline ---
34query = eta_calculated.writeStream \
35    .foreachBatch(write_etas_to_dynamodb) \
36    .trigger(processingTime="30 seconds") \
37    .outputMode("update") \
38    .start()
39
40print("Pipeline ETA iniciado -- trigger cada 30 segundos")
41query.awaitTermination()

Sink a DynamoDB con foreachBatch -- el frontend de Ana lee de aquí

foreachBatch es la forma correcta de escribir a sistemas externos desde Spark Streaming. No uses foreach (procesa fila a fila, extremadamente lento). Con foreachBatch agrupas escrituras y puedes usar batch_writer que envía hasta 25 items por request HTTP. El batch_id te da idempotencia si necesitas reintentos: puedes verificar si un batch ya se procesó antes de escribirlo.

### El campo is_at_risk -- la alerta que pide Ana

Para implementar el requisito de Ana, necesitas comparar la hora estimada de llegada (ahora + eta_minutes) con la hora prometida al cliente (promised_eta). Si la estimación supera la promesa, la entrega está "at risk" y el coordinador debe actuar.

1from pyspark.sql.functions import (
2    unix_timestamp, to_timestamp, when, col, current_timestamp, lit
3)
4
5# Anadir is_at_risk y minutes_over_sla al DataFrame
6eta_with_risk = eta_calculated \
7    .withColumn(
8        "estimated_arrival_ts",
9        unix_timestamp(current_timestamp()) + col("eta_minutes") * lit(60)
10    ) \
11    .withColumn(
12        "promised_ts",
13        unix_timestamp(to_timestamp(col("promised_eta")))
14    ) \
15    .withColumn(
16        "is_at_risk",
17        col("estimated_arrival_ts") > col("promised_ts")
18    ) \
19    .withColumn(
20        "minutes_over_sla",
21        when(col("is_at_risk"),
22             (col("estimated_arrival_ts") - col("promised_ts")) / lit(60)
23        ).otherwise(lit(0))
24    ) \
25    .withColumn(
26        "risk_level",
27        when(col("minutes_over_sla") > 15, lit("critical"))
28        .when(col("minutes_over_sla") > 5, lit("high"))
29        .when(col("is_at_risk"), lit("medium"))
30        .otherwise(lit("ok"))
31    )
32
33# El frontend interpreta risk_level como:
34# "ok" -> verde, "medium" -> amarillo, "high" -> naranja, "critical" -> rojo

Cálculo de is_at_risk y clasificación por niveles de riesgo

Con esto, tu primer día tiene un pipeline funcional end-to-end: GPS -> Kafka -> Spark (Haversine + ETA + risk) -> DynamoDB -> Dashboard de Ana. A las 18:00 lo deploys en el cluster de EMR, verificas que los ETAs aparecen en DynamoDB y escribes a Ana por Slack: "ETAs live en el dashboard. Refresh cada 30 segundos. Mañana agrego el dashboard de SLAs." Respuesta de Ana: "Ok." Una palabra. Suficiente.

## ejercicios

[01]

Implementar cálculo de distancia Haversine

Implementa la función haversine_km que calcula la distancia entre dos puntos GPS. Luego crea estimate_eta que use la distancia, velocidad y un factor urbano para estimar minutos restantes. Valida con los datos de ejemplo de FlashRoute.

Cargando editor...

[02]

Detectar entregas en riesgo de incumplir SLA

Implementa la logica que determina si una entrega va a llegar tarde comparando el ETA calculado con el promised_eta. Clasifica en verde (on_time), amarillo (at_risk) y rojo (late). Luego procesa una flota completa y calcula el porcentaje de cumplimiento.

Cargando editor...

[03]

Pipeline Spark Streaming: GPS a ETA

Escribe el job completo de Spark Structured Streaming que lee eventos GPS de Kafka, calcula el ETA usando Haversine y prepara el DataFrame final con is_at_risk para escribir a DynamoDB. El pipeline debe procesar cada 30 segundos.

Cargando editor...

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...