Estado

Un pago generado a las 10:03 puede llegar a las 10:17 por una desconexión móvil. Agrupar por la hora de ingesta lo asignaría a una ventana distinta y haría que un reproceso produzca otro resultado. La columna de event time debe proceder del evento, convertirse a `timestamp` y validarse antes de cualquier operación temporal.

El retraso `processing_time - event_time` es una distribución, no una constante. Para elegir tolerancia se estudian percentiles y casos extremos por fuente. Un reloj del productor defectuoso debe enviarse a cuarentena; ampliar indefinidamente el watermark para ocultarlo traslada el problema al state store.

from pyspark.sql import functions as F

events = (
    raw.withColumn("event_ts", F.to_timestamp("event_time_iso"))
       .withColumn("ingested_at", F.current_timestamp())
       .withColumn(
           "lateness_seconds",
           F.col("ingested_at").cast("long") - F.col("event_ts").cast("long")
       )
)

valid = events.where("event_ts IS NOT NULL AND lateness_seconds >= 0")

Las ventanas tumbling no se solapan; las sliding pueden asignar el mismo evento a varias ventanas. En una agregación, Spark mantiene acumuladores para ventanas todavía abiertas. El watermark avanza a partir del máximo event time observado menos el retraso configurado y permite retirar estado antiguo.

Una tolerancia corta reduce memoria y acelera resultados finales, pero descarta más eventos tardíos. Una tolerancia larga mejora completitud a costa de latencia y estado. La decisión debe expresar un compromiso medible, por ejemplo aceptar el 99,5% de eventos en quince minutos y corregir el resto mediante un proceso separado.

from pyspark.sql import functions as F

sales_5m = (
    events.withWatermark("event_ts", "15 minutes")
      .groupBy(F.window("event_ts", "5 minutes"), "store_id")
      .agg(
          F.countDistinct("order_id").alias("orders"),
          F.sum("amount").alias("revenue")
      )
)

`dropDuplicatesWithinWatermark(["event_id"])` recuerda las claves observadas mientras el evento pueda seguir teniendo duplicados dentro de la tolerancia. La marca temporal debe definirse antes. Si dos copias pueden diferir en timestamp, la tolerancia debe superar la máxima distancia temporal entre ellas para garantizar la deduplicación.

La clave debe representar identidad de negocio, no una combinación accidental de todas las columnas. Antes de deduplicar conviene validar nulos y normalizar tipos. Los eventos que llegan más tarde que el umbral pueden descartarse; la métrica de descarte y una ruta de reconciliación son parte del diseño.

deduplicated = (
    events.where("event_id IS NOT NULL AND event_ts IS NOT NULL")
      .withWatermark("event_ts", "30 minutes")
      .dropDuplicatesWithinWatermark(["event_id"])
)

Relacionar clics con pagos solo por `session_id` deja abierta para siempre la posibilidad de una coincidencia futura. Añadir que el pago ocurra entre el clic y treinta minutos después proporciona un intervalo acotado. Cada entrada debe declarar su tolerancia de tardanza para que el motor calcule cuándo retirar estado.

Los joins outer requieren esperar a que transcurra el horizonte antes de emitir una fila no emparejada. Eso aumenta la latencia de los `NULL` esperados. En múltiples streams, la política global por defecto avanza con el watermark más lento; cambiarla a `max` reduce latencia, pero puede descartar datos del stream rezagado.

clicks_wm = clicks.withWatermark("click_ts", "20 minutes")
payments_wm = payments.withWatermark("payment_ts", "10 minutes")

matched = clicks_wm.join(
    payments_wm,
    (clicks_wm.session_id == payments_wm.session_id) &
    (payments_wm.payment_ts >= clicks_wm.click_ts) &
    (payments_wm.payment_ts <= clicks_wm.click_ts + F.expr("INTERVAL 30 MINUTES")),
    "leftOuter",
)

`stateOperators` informa filas totales y actualizadas, memoria usada y filas retiradas por watermark. Un crecimiento continuo después de varios horizontes puede indicar claves de cardinalidad extrema, timestamps futuros, ausencia de watermark efectivo o una condición de join no acotada.

El diagnóstico debe reproducir datos puntuales: a tiempo, duplicados, tardíos dentro de tolerancia, demasiado tardíos y timestamps inválidos. Una prueba de recuperación reinicia desde el mismo checkpoint; cambiar access mode o esquema de estado durante el experimento puede introducir otra variable.

progress = query.lastProgress or {}
for operator in progress.get("stateOperators", []):
    print({
        "rows_total": operator.get("numRowsTotal"),
        "rows_updated": operator.get("numRowsUpdated"),
        "rows_removed": operator.get("numRowsRemoved"),
        "memory_bytes": operator.get("memoryUsedBytes"),
    })