Joins, agregaciones y Watermarks en Streaming

Objetivo: entender por qué Spark necesita guardar datos temporalmente (state), cómo combinar streams y cómo evitar que el estado crezca indefinidamente.

1. ¿Qué es el estado (State)?

Imagina que queremos calcular las ventas de cada farmacia cada 5 minutos.

Llegan tres tickets:

Spark tiene que recordar las cantidades anteriores mientras espera nuevos eventos.

Ese acumulador de 225 € forma parte del estado.

El problema: si tenemos millones de farmacias, clientes o ventanas temporales, el estado puede crecer muchísimo.

Aquí es donde entran los watermarks.

2. ¿Qué es un Watermark?

Un watermark indica hasta qué punto del tiempo de los eventos podemos considerar que los datos antiguos han llegado.

Por ejemplo:

.withWatermark("event_time", "10 minutes")

Significa que permitimos un retraso de referencia de 10 minutos respecto al avance del tiempo de evento observado.

Simplificando:

$Watermark\approx Max\left(event\_time\right)-10minutos$Watermark≈Max(event_time)−10 minutos

Si el evento más reciente tiene hora 10:30, el watermark de referencia sería 10:20.

En una agregación por ventanas, Spark puede cerrar y liberar el estado de aquellas ventanas cuya finalización queda por detrás del watermark. Un evento demasiado tardío puede quedar descartado. Databricks on AWS

3. Simulador: ¿qué ocurre con un ticket que llega tarde?

La gran decisión: cuanto mayor sea la tolerancia, más eventos tardíos podemos admitir, pero también tendremos que conservar estado durante más tiempo.

4. Código real: ventas cada 5 minutos

Vamos a crear una Streaming Table que calcula las ventas por farmacia y ventana temporal.

from pyspark import pipelines as dp
from pyspark.sql import functions as F

@dp.table(name="sales_5min")
def sales_5min():

    tickets = (
        spark.readStream.table("silver_tickets")
        .withWatermark("event_time", "10 minutes")
    )

    return (
        tickets
        .groupBy(
            F.window("event_time", "5 minutes"),
            F.col("pharmacy_id")
        )
        .agg(
            F.sum("amount").alias("total_sales")
        )
    )

Tres ideas:

• withWatermark establece cuánto retraso admitimos.

• window agrupa los tickets en intervalos de 5 minutos.

• groupBy + sum mantiene el estado necesario para sumar los eventos.

Importante: con una agregación streaming en modo append, los resultados de una ventana se publican cuando esta se cierra según el watermark, no necesariamente cada 5 minutos. Para un KPI operacional que deba mostrar importes parciales inmediatamente, este diseño puede no ser el apropiado.

Databricks recomienda watermarks en las agregaciones streaming para controlar el estado y mantener el procesamiento incremental. Databricks on AWS

5. Joins: dos escenarios diferentes

A. Stream + tabla estática

Queremos enriquecer cada ticket con el nombre de su farmacia.

@dp.table(name="enriched_tickets")
def enriched_tickets():

    tickets = spark.readStream.table("silver_tickets")
    pharmacies = spark.read.table("dim_pharmacy")

    return tickets.join(
        pharmacies,
        "pharmacy_id",
        "left"
    )

Es un stream-static join. Normalmente no necesitamos watermark para este caso.

Pero hay un matiz: si mañana cambia el nombre de una farmacia, los tickets ya procesados no se actualizan automáticamente. El join se aplica cuando se procesa cada microbatch. Databricks on AWS

B. Stream + Stream

Imagina dos streams:

• tickets: compras realizadas.

• app_tracking: interacciones del cliente con la aplicación.

Queremos relacionar una compra con interacciones que ocurrieron durante los 10 minutos anteriores.

tickets = (
    spark.readStream.table("silver_tickets")
    .withWatermark("event_time", "15 minutes")
)

tracking = (
    spark.readStream.table("app_tracking")
    .withWatermark("event_time", "15 minutes")
)

result = tickets.alias("t").join(
    tracking.alias("a"),
    """
    t.sso_id = a.sso_id
    AND a.event_time BETWEEN
        t.event_time - INTERVAL 10 MINUTES
        AND t.event_time
    """,
    "inner"
)

Aquí tenemos dos condiciones fundamentales:

1. Watermarks en ambos streams: delimitan cuánto estado necesitamos conservar.

2. Condición temporal del join: especifica hasta cuándo dos eventos pueden coincidir.

Sin estos límites, Spark podría necesitar conservar una enorme cantidad de eventos antiguos buscando posibles coincidencias. Databricks on AWS

6. ¿Qué elegiría para vuestra plataforma?

Para los KPIs diarios que mostráis en Power BI, priorizaría una Materialized View si necesitáis incorporar correcciones históricas y mantener resultados actualizados, en lugar de perder tickets que llegan tarde por un watermark demasiado pequeño.

7. Mini ejercicio

Qué debes recordar:

• State: memoria lógica de eventos anteriores necesaria para calcular resultados.

• Watermark: controla cuándo pueden descartarse estados antiguos y eventos demasiado tardíos.

• Stream-static join: relativamente sencillo, pero no corrige automáticamente resultados históricos.

• Stream-stream join: necesita una estrategia explícita para limitar el estado.

La siguiente lección será especialmente útil para evaluar Lakeflow frente a dbt: cómo funcionan realmente las actualizaciones incrementales y cuándo Spark tiene que recalcular datos.