Kafka

El DataFrame de Kafka expone `key`, `value`, `topic`, `partition`, `offset`, `timestamp` y headers. `key` y `value` llegan como bytes, por lo que hacer solo `CAST(value AS STRING)` no valida el contrato. Un esquema explícito con `from_json` permite distinguir una fila corrupta de una evolución compatible.

Conservar `topic`, `partition` y `offset` en bronze permite investigar pérdidas, reconstruir un rango y demostrar qué mensaje originó una fila. La clave Kafka también importa: el productor la usa para particionar y preservar orden dentro de una partición, pero Kafka no ofrece orden global entre particiones.

kafka_raw = (
    spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", bootstrap_servers)
      .option("subscribe", "orders.v1")
      .load()
)

bronze = kafka_raw.selectExpr(
    "CAST(key AS STRING) AS message_key",
    "CAST(value AS STRING) AS payload",
    "topic", "partition", "offset",
    "timestamp AS kafka_timestamp"
)

`subscribe` elige topics concretos, `subscribePattern` usa una expresión regular y `assign` fija particiones. Debe configurarse exactamente uno. En streaming, `startingOffsets` es `latest` por defecto y solo se consulta si no existe progreso previo; cambiarlo después no rebobina una consulta con checkpoint.

Para un backfill se usa un checkpoint y destino separados o una lectura batch con rangos explícitos, no se modifica a ciegas una consulta productiva. `failOnDataLoss=false` permite continuar cuando offsets ya no existen, pero acepta una posible pérdida y debe acompañarse de reconciliación; no es una solución genérica para errores.

orders = (
    spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", bootstrap_servers)
      .option("subscribe", "orders.v1")
      .option("startingOffsets", "earliest")
      .option("failOnDataLoss", "true")
      .option("maxOffsetsPerTrigger", 500000)
      .load()
)

Con JSON, un `StructType` explícito evita inferencia por microbatch y permite detectar `_corrupt_record` o campos obligatorios nulos. Avro o Protobuf con un schema registry ofrecen contratos más fuertes, pero siguen necesitando una política de compatibilidad entre productor y consumidor.

Un patrón robusto publica el mensaje crudo en bronze, enruta fallos a cuarentena con el motivo y solo expone columnas tipadas en silver. Añadir un campo opcional suele ser compatible; renombrar o cambiar tipo requiere una migración coordinada o lectura multi-versión mediante un campo `schema_version`.

from pyspark.sql import functions as F, types as T

order_schema = T.StructType([
    T.StructField("order_id", T.StringType(), False),
    T.StructField("event_ts", T.TimestampType(), False),
    T.StructField("amount", T.DecimalType(18, 2), False),
    T.StructField("schema_version", T.IntegerType(), False),
])

decoded = bronze.withColumn(
    "event", F.from_json("payload", order_schema)
)

Spark registra rangos de offsets en el checkpoint y Delta confirma cada microbatch de manera transaccional. Un fallo puede hacer que el motor vuelva a calcular el lote, pero el protocolo del sink evita materializarlo dos veces. Esto no deduplica dos mensajes distintos que el productor publicó con el mismo `order_id`.

Cuando el destino es una API, una base no transaccional o varios sistemas, `foreachBatch` ofrece semántica at-least-once a menos que la función sea idempotente. Un outbox del productor, identificadores de evento estables y `MERGE` en silver resuelven capas distintas del problema.

MERGE INTO main.silver.orders AS target
USING staged_orders AS source
ON target.order_id = source.order_id
WHEN MATCHED AND source.event_ts > target.event_ts THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *;

El paralelismo máximo de lectura está condicionado por las particiones Kafka, aunque Spark pueda dividir rangos grandes en más tareas según capacidades de la fuente. `maxOffsetsPerTrigger` limita el volumen de un microbatch y protege al sink durante recuperación, pero si queda por debajo de la tasa de llegada el lag crecerá indefinidamente.

Las métricas `avgOffsetsBehindLatest`, `maxOffsetsBehindLatest` y bytes estimados muestran atraso por fuente. Deben correlacionarse con duración del trigger, distribución de particiones y tasas de procesamiento. Una sola partición caliente puede dominar el SLA aunque la media parezca saludable.

progress = query.lastProgress or {}
for source in progress.get("sources", []):
    metrics = source.get("metrics", {})
    print({
        "description": source.get("description"),
        "avg_lag": metrics.get("avgOffsetsBehindLatest"),
        "max_lag": metrics.get("maxOffsetsBehindLatest"),
        "bytes_behind": metrics.get("estimatedTotalBytesBehindLatest"),
    })