Streaming

Una lectura con `readStream` describe un flujo que todavía no está en ejecución. Spark construye el plan, descubre qué datos nuevos están disponibles en cada trigger y solo entonces materializa un microbatch. La misma API de DataFrames permite reutilizar transformaciones batch, pero una operación válida en batch puede ser inviable en streaming si exige conservar estado sin límite.

En un pipeline de pedidos conviene separar tres decisiones: cómo se descubre la entrada, qué transformaciones son incrementales y cómo se confirma la salida. El SLA no lo decide `readStream`; lo determinan el trigger, el volumen por lote, la capacidad y el comportamiento del sink ante reintentos.

orders = spark.readStream.table("main.bronze.orders")

query = (
    orders.where("order_id IS NOT NULL")
      .writeStream
      .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_silver")
      .trigger(processingTime="1 minute")
      .toTable("main.silver.orders")
)

`processingTime` mantiene una consulta activa y lanza microbatches con la periodicidad solicitada, siempre que el anterior haya acabado. Es apropiado para un dashboard que debe actualizarse durante todo el día. Reducir el intervalo por debajo de la duración real del lote solo aumenta la presión de planificación.

`availableNow=True` procesa todos los datos disponibles en uno o varios microbatches y finaliza. Encaja con trabajos incrementales programados porque conserva checkpoints y límites por lote sin pagar por una consulta ociosa. También simplifica backfills controlados: el trabajo termina cuando alcanza el inicio del trigger.

(
  spark.readStream.table("main.bronze.order_events")
    .writeStream
    .option("checkpointLocation", "/Volumes/main/ops/checkpoints/order_events")
    .trigger(availableNow=True)
    .toTable("main.silver.order_events")
    .awaitTermination()
)

Al reiniciar, Spark consulta el checkpoint para conocer qué rangos de la fuente fueron procesados y qué lotes se confirmaron en el sink. Las operaciones con estado añaden metadatos y datos del state store. Borrar la carpeta convierte el siguiente arranque en una consulta nueva y puede reintroducir datos o perder la posición esperada.

No todos los cambios de código son compatibles con un checkpoint existente. Cambiar el número de fuentes, el topic Kafka, el tipo de sink, las claves de una agregación o el esquema del estado suele exigir un checkpoint nuevo y un plan de migración o reproceso. Un simple filtro suele ser compatible, aunque puede cambiar el resultado funcional.

catalog = spark.conf.get("app.catalog", "main")
environment = spark.conf.get("app.environment", "dev")
checkpoint = f"/Volumes/{catalog}/ops/checkpoints/{environment}/orders_v1"

assert environment in {"dev", "test", "prod"}
print({"checkpoint": checkpoint, "query_version": 1})

Los sinks Delta integrados coordinan los commits de cada microbatch con el checkpoint. Si una tarea falla después de escribir pero antes de confirmar, el reinicio puede volver a presentar el mismo lote y el sink debe reconocerlo. La garantía de la fuente por sí sola no impide duplicados en un servicio externo.

`foreachBatch` abre la puerta a `MERGE`, múltiples destinos o APIs, pero transfiere responsabilidad al código. El `batch_id` y una clave de negocio estable permiten registrar el lote o hacer upsert determinista. Enviar eventos a un endpoint sin clave idempotente conserva, como mucho, semántica at-least-once.

from delta.tables import DeltaTable

def upsert_orders(batch_df, batch_id):
    target = DeltaTable.forName(spark, "main.silver.orders_current")
    (target.alias("t")
      .merge(batch_df.alias("s"), "t.order_id = s.order_id")
      .whenMatchedUpdateAll()
      .whenNotMatchedInsertAll()
      .execute())

(orders.writeStream
  .foreachBatch(upsert_orders)
  .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_upsert")
  .start())

`lastProgress` expone tasas de entrada y proceso, duración de fases, offsets y métricas de estado. Si `inputRowsPerSecond` supera sostenidamente `processedRowsPerSecond`, crece el backlog; si no hay filas nuevas, una latencia alta puede proceder del trigger o del propio sink. En Kafka, el desfase de offsets aporta otra señal independiente.

La operación necesita umbrales vinculados al SLA: frescura máxima, duración p95 del lote, filas descartadas y tamaño del estado. Un runbook útil conecta cada alerta con hipótesis y acciones seguras, evitando reiniciar o borrar checkpoints sin diagnóstico.

progress = query.lastProgress or {}
summary = {
    "batch_id": progress.get("batchId"),
    "input_rps": progress.get("inputRowsPerSecond", 0),
    "processed_rps": progress.get("processedRowsPerSecond", 0),
    "batch_ms": progress.get("durationMs", {}).get("triggerExecution"),
    "state": progress.get("stateOperators", []),
}
print(summary)