Puente Batch-Streaming: Un Solo Lago para Ambos Pipelines

Visión General

Tu ETL nocturno acaba de cargar los números de ayer en el warehouse — y un cliente en el dashboard del equipo de fraude está reclamando un cargo de hace cuarenta segundos. Para el warehouse, ese pedido todavía no existe. Existirá mañana a las 2 a.m.

El puente batch-to-streaming cierra esa brecha. Los pipelines batch procesan el histórico en bloques programados; los pipelines streaming persiguen eventos conforme ocurren. El puente hace que ambas rutas escriban registros compatibles en el mismo data lake particionado, de modo que la capa de servicio pueda responder “todo lo que sabemos de este cliente” sin importar qué ruta produjo cada fila. Es la arquitectura Lambda hecha concreta: una capa batch para completitud, una capa speed para frescura, fusionadas en el momento de la consulta. La parte difícil nunca es la fontanería; es mantener idénticas las reglas de esquema, particionado y deduplicación en dos pipelines que evolucionan por separado.

Cuándo Usarlo

• Analítica que necesita eventos frescos e historia profunda en una sola consulta: scoring de fraude, dashboards en vivo sobre años de datos.
• Migración incremental de ETL batch a streaming: correr ambos en paralelo y dejar que la capa de servicio oculte la transición.
• Sistemas alimentados por CDC donde la misma entidad llega por un topic de Kafka y por una carga masiva nocturna.
• Data lakes donde los backfills batch y los appends streaming deben coexistir en las mismas tablas.

Cuándo No Usarlo

• Sistemas puramente streaming: corre Kappa (abajo) o Kafka + Flink directamente, en vez de mantener una ruta batch que no necesitas.
• Sistemas puramente batch donde una latencia de horas es aceptable: una capa speed es deuda operativa sin beneficio alguno.
• Equipos pequeños que no pueden costear dos rutas de código. El coste honesto de Lambda es que cada cambio de esquema, cada regla de dedup y cada decisión de particionado se implementa dos veces.
• Organizaciones que pueden adoptar un formato lakehouse (Delta Lake, Apache Iceberg): los upserts ACID te dan casi todo este patrón con un solo pipeline.

Solución

Dos rutas, un lago. La ruta streaming aterriza eventos en S3 a través de Kafka Connect; la ruta batch aterriza cargas masivas sobre el mismo esquema de particiones; Trino consulta la unión y deduplica; Redis sirve lecturas de menos de un segundo para el estado más fresco.

Kafka Connect S3 sink para la ruta streaming

{
  "name": "s3-customers-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "customer-events",
    "s3.bucket.name": "data-lake-shop",
    "s3.region": "us-east-1",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": "3600000",
    "path.format": "year=YYYY/month=MM/day=dd/hour=HH",
    "locale": "en_US",
    "timezone": "UTC",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "schema.compatibility": "BACKWARD",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.enhanced.avro.schema.support": "true"
  }
}

Los dos campos que importan para el puente son path.format, que decide el layout de particiones que el resto de rutas debe igualar, y schema.compatibility, que el registro de esquemas compartido hace cumplir en ambas rutas.

Ruta batch escribiendo al mismo data lake

# batch_to_lake.py — ETL batch escribiendo al mismo path de S3 que streaming
import boto3
import pandas as pd
from datetime import datetime

class BatchToLake:
    def __init__(self, bucket, prefix):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.prefix = prefix

    def load_batch(self, df, table_name, batch_hour):
        """Escribe datos batch con el mismo formato de path que streaming."""
        # Igualar el formato de partición de streaming exactamente, hora incluida:
        # year=YYYY/month=MM/day=dd/hour=HH
        parts = batch_hour.strftime('%Y/%m/%d/%H').split('/')
        year, month, day, hour = parts

        s3_key = (
            f"{self.prefix}/{table_name}"
            f"/year={year}/month={month}/day={day}/hour={hour}"
            f"/batch_{batch_hour.strftime('%Y%m%d%H')}.parquet"
        )

        parquet_buffer = df.to_parquet(engine='pyarrow', compression='snappy')
        self.s3.put_object(Bucket=self.bucket, Key=s3_key, Body=parquet_buffer)
        return s3_key

    def list_partitions(self, table_name, date):
        """Lista los ficheros de una fecha — batch y streaming aterrizan aquí."""
        year, month, day = date.strftime('%Y'), date.strftime('%m'), date.strftime('%d')
        prefix = f"{self.prefix}/{table_name}/year={year}/month={month}/day={day}/"

        response = self.s3.list_objects_v2(Bucket=self.bucket, Prefix=prefix)
        return [
            {
                'key': obj['Key'],
                'size': obj['Size'],
                'source': 'batch' if 'batch_' in obj['Key'] else 'streaming'
            }
            for obj in response.get('Contents', [])
        ]

El prefijo batch_ en el nombre de fichero es deliberado: permite a la capa de servicio etiquetar _source sin inspeccionar metadatos, y hace trivial depurar el contenido de una partición. El segmento hour importa más de lo que parece; quítalo y los ficheros batch caen fuera de las particiones declaradas de la tabla, invisibles para Trino.

Capa de servicio unificada con Trino

-- serving_layer.sql — consulta datos batch y streaming juntos
-- Ambas rutas escriben a la misma tabla particionada en S3;
-- Trino las trata como una sola tabla.

CREATE EXTERNAL TABLE IF NOT EXISTS lake.customers (
    id BIGINT,
    email VARCHAR,
    status VARCHAR,
    is_active BOOLEAN,
    created_at TIMESTAMP,
    _source VARCHAR,
    _ingested_at TIMESTAMP
)
STORED AS PARQUET
LOCATION 's3://data-lake-shop/customers/'
PARTITIONED BY (
    year VARCHAR,
    month VARCHAR,
    day VARCHAR,
    hour VARCHAR
);

-- Descubre particiones escritas desde la última reparación
MSCK REPAIR TABLE lake.customers;

-- Una consulta sobre ambas rutas
SELECT
    _source,
    COUNT(*) AS row_count,
    MIN(_ingested_at) AS earliest,
    MAX(_ingested_at) AS latest
FROM lake.customers
WHERE year = '2026' AND month = '08' AND day = '19'
GROUP BY _source;

-- Fusiona batch y streaming en una vista deduplicada:
-- prefiere streaming (más fresco), cae a batch si falta
CREATE OR REPLACE VIEW warehouse.customers_unified AS
SELECT
    id,
    email,
    status,
    is_active,
    created_at,
    ROW_NUMBER() OVER (
        PARTITION BY id
        ORDER BY
            CASE _source WHEN 'streaming' THEN 1 WHEN 'batch' THEN 2 END,
            _ingested_at DESC
    ) AS rn
FROM lake.customers
WHERE year = CAST(YEAR(CURRENT_DATE) AS VARCHAR)
  AND month = LPAD(CAST(MONTH(CURRENT_DATE) AS VARCHAR), 2, '0');

CREATE OR REPLACE VIEW warehouse.customers_latest AS
SELECT id, email, status, is_active, created_at
FROM warehouse.customers_unified
WHERE rn = 1;

La regla de dedup codifica el supuesto central del patrón: las filas streaming son más frescas y las batch son la red de seguridad. El CASE _source ordena streaming primero, y _ingested_at DESC desempata dentro de cada fuente.

Procesador Flink para la capa speed

// StreamingProcessor.java — procesador Flink para la capa speed
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.formats.avro.AvroDeserializationSchema;
import org.apache.flink.formats.avro.AvroSerializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import java.time.Duration;

public class StreamingProcessor {
    public static void main(String[] args) throws Exception {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<Customer> source = KafkaSource.<Customer>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("customer-events")
            .setGroupId("flink-processor")
            .setValueOnlyDeserializer(
                AvroDeserializationSchema.forSpecific(Customer.class)
            )
            .build();

        DataStream<Customer> events = env.fromSource(
            source,
            // Desorden acotado: eventos con más de 5 min de retraso
            // van a la salida lateral, no al stream principal
            WatermarkStrategy.<Customer>forBoundedOutOfOrderness(Duration.ofMinutes(5)),
            "customer-events"
        );

        DataStream<CustomerEnriched> enriched = events
            .map(customer -> {
                var e = new CustomerEnriched();
                e.setId(customer.getId());
                e.setEmail(customer.getEmail().toLowerCase());
                e.setStatus(customer.getStatus());
                e.setActive("active".equals(customer.getStatus()));
                e.setProcessedAt(System.currentTimeMillis());
                return e;
            })
            .name("enrich");

        // Consumidores en tiempo real tras Kafka
        KafkaSink<CustomerEnriched> kafkaSink = KafkaSink.<CustomerEnriched>builder()
            .setBootstrapServers("kafka:9092")
            .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                    .setTopic("customer-events-enriched")
                    .setValueSerializationSchema(
                        AvroSerializationSchema.forSpecific(CustomerEnriched.class)
                    )
                    .build()
            )
            .build();

        enriched.sinkTo(kafkaSink).name("kafka-sink");

        // Capa speed: último estado por clave para lecturas sub-segundo
        enriched.addSink(new RedisSink<>()).name("redis-sink");

        env.execute("customer-stream-processor");
    }
}

Fíjate en los watermarks: forBoundedOutOfOrderness(5 minutos) es la mitad streaming de la política de datos tardíos. Los eventos que llegan después del watermark se enrutan a una salida lateral (o se descartan); la ruta batch los recoge en la siguiente ejecución. Eso es el puente funcionando como debe: el dato tardío no desaparece, solo entra por la puerta más lenta.

Consumidor unificado en Python

# unified_consumer.py — historia desde Trino, estado fresco desde la capa speed
import json
import trino
import pandas as pd
import redis

class UnifiedConsumer:
    def __init__(self, trino_host, redis_client):
        self.trino = trino.dbapi.connect(host=trino_host, port=443, user="analyst")
        self.redis = redis_client  # la capa speed a la que escribe Flink

    def get_historical(self, customer_id):
        """Historia deduplicada de la capa batch vía Trino."""
        query = "SELECT * FROM warehouse.customers_latest WHERE id = ?"
        return pd.read_sql(query, self.trino, params=[customer_id])

    def get_realtime(self, customer_id):
        """Último estado desde la capa speed (lookup por clave, no scan)."""
        raw = self.redis.get(f"customer:{customer_id}")
        return json.loads(raw) if raw else None

    def get_unified_view(self, customer_id):
        """Combina la historia batch con el estado streaming más fresco."""
        historical = self.get_historical(customer_id)
        result = historical.iloc[0].to_dict() if not historical.empty else {}

        latest = self.get_realtime(customer_id)
        if latest:
            # Streaming sobrescribe solo los campos que realmente trae
            for field in ('email', 'status', 'is_active'):
                if field in latest:
                    result[field] = latest[field]
            result['_source'] = 'merged'

        return result or None

Dos cosas cambian frente a una implementación ingenua. La lectura de la capa speed es un GET por clave, no un consumidor de Kafka; re-consumir el topic para encontrar el último evento de un cliente barrería el tráfico de los demás y bloquearía. Y la fusión aplica los campos streaming de forma selectiva, para que una actualización parcial no pueda borrar campos que no trae.

Alineación de esquemas entre batch y streaming

# schema_alignment.py — un solo modelo que ambas rutas deben producir
from pydantic import BaseModel
from datetime import datetime
from typing import Literal

class CustomerRecord(BaseModel):
    """Esquema compartido por las rutas batch y streaming."""
    id: int
    email: str
    status: Literal["active", "inactive", "banned", "suspended"]
    is_active: bool
    created_at: datetime
    _source: Literal["batch", "streaming"]
    _ingested_at: datetime

class BatchProducer:
    def produce(self, row):
        return CustomerRecord(
            id=row['id'],
            email=row['email'].lower().strip(),
            status=row['status'],
            is_active=row['status'] == 'active',
            created_at=row['created_at'],
            _source="batch",
            _ingested_at=datetime.now()
        )

class StreamingProducer:
    def produce(self, event):
        # Envelope CDC: la fila vive bajo 'after'
        after = event['after']
        return CustomerRecord(
            id=after['id'],
            email=after['email'].lower().strip(),
            status=after['status'],
            is_active=after['status'] == 'active',
            created_at=after['created_at'],
            _source="streaming",
            _ingested_at=datetime.now()
        )

Ambos productores ejecutan la misma normalización: email en minúsculas, is_active derivado, _source etiquetado. Si una ruta pasa a minúsculas y la otra no, el dedup por id sigue funcionando pero cada filtro posterior sobre email se divide en silencio. En producción el schema registry lo impone en el cable; el modelo compartido lo impone en el código.

Variantes

Arquitectura Kappa (solo streaming)

# kappa_architecture.py — sustituir batch por streaming re-procesado
from kafka import KafkaConsumer
from datetime import datetime

class KappaArchitecture:
    """Sin capa batch. Todo fluye por streaming; el re-procesado
    histórico re-lee el topic de Kafka desde el principio."""
    def reprocess(self, kafka_topic, new_processor):
        # Un consumer group nuevo arranca en el offset más antiguo
        consumer = KafkaConsumer(
            kafka_topic,
            auto_offset_reset='earliest',
            enable_auto_commit=False,
            group_id=f'reprocess-{datetime.now().timestamp()}'
        )

        for message in consumer:
            new_processor.process(message.value)

Kappa elimina la segunda ruta de código por completo, pero solo funciona si la retención cubre tu historia o si el replay con compaction es fiel, y si “re-procesar” no exige un modelo de cómputo distinto al del procesamiento en vivo. Los equipos eligen Lambda no porque sea elegante sino porque sus jobs batch hacen de verdad algo que streaming no puede: joins contra snapshots, entrenamiento de modelos, conciliación regulatoria.

Formato lakehouse (la respuesta moderna)

Delta Lake y Apache Iceberg merecen mención porque colapsan buena parte de este patrón en la capa de almacenamiento: los upserts ACID permiten que escritores streaming y batch actualicen la misma tabla sin convenciones de partición, etiquetas _source ni dedup en consulta; el motor fusiona por clave primaria. Si empiezas de cero sobre un lakehouse, puede que no necesites el puente en absoluto. El patrón sigue vigente en lagos Parquet planos, en tablas Hive existentes y en cualquier sitio donde el coste de migración descarte cambiar de formato.

Híbrido con vistas materializadas

-- materialized_view.sql — precomputar la fusión para consultas calientes
CREATE MATERIALIZED VIEW warehouse.customer_stats_hourly AS
SELECT
    DATE_TRUNC('hour', _ingested_at) AS hour,
    _source,
    COUNT(*) AS event_count,
    COUNT(DISTINCT id) AS unique_customers,
    SUM(CASE WHEN is_active THEN 1 ELSE 0 END) AS active_count
FROM lake.customers
WHERE year = CAST(YEAR(CURRENT_DATE) AS VARCHAR)
  AND month = LPAD(CAST(MONTH(CURRENT_DATE) AS VARCHAR), 2, '0')
GROUP BY DATE_TRUNC('hour', _ingested_at), _source;

REFRESH MATERIALIZED VIEW warehouse.customer_stats_hourly;

Qué Funciona

• Usa un schema registry compartido — ambas rutas deben producir el mismo esquema Avro/Protobuf, validado en escritura. Mira el patrón de evolución de esquemas para las reglas de compatibilidad.
• Etiqueta los registros con _source — distingue batch de streaming en cada consulta; además es tu primera herramienta de depuración cuando los recuentos no cuadran.
• Iguala los formatos de partición exactamente — ambas rutas escriben year=YYYY/month=MM/day=dd/hour=HH. Un segmento que falte y la capa de servicio lee la mitad de los datos.
• Haz ambas rutas idempotentes — los re-runs y backfills son rutina; las claves de dedup (id + _ingested_at) deben sobrevivirlos.
• Monitoriza el lag en ambos lados — lag streaming significa capa speed sirviendo datos viejos; lag batch significa huecos en la historia. Alerta en los dos.
• Prefiere Kappa o un formato lakehouse si empiezas de cero — mantener dos pipelines es el coste real de Lambda.

Errores Comunes

• Deriva de esquema entre rutas: batch pasa email a minúsculas, streaming no; los filtros se dividen en silencio. Impone un único modelo compartido.
• Desajuste de particiones: batch escribe granularidad day= mientras la tabla espera hour=; los ficheros batch se vuelven invisibles para el motor. Alinea los formatos hasta el último segmento.
• Sin dedup en la capa de servicio: un cliente aparece dos veces, una por ruta. ROW_NUMBER() sobre id ordenado por frescura de fuente lo arregla.
• Leer estado fresco re-consumiendo Kafka: barrer el topic buscando una clave bloquea y se traga los eventos de las demás. La capa speed existe para lookups por clave — úsala.
• Solo streaming sin fallback batch: cuando el pipeline de stream falle (y fallará), no hay nada contra lo que conciliar. Mantén la ruta batch aunque sea lenta.

Lecturas Adicionales

• Confluent: S3 Sink Connector — particionado, formatos e integración con schema registry para la ruta de escritura streaming.
• AWS Prescriptive Guidance: Build a Lambda architecture — la versión con servicios gestionados de este patrón.
• Documentación de Trino: conector Hive — tablas externas, particiones y semántica de MSCK REPAIR.
• Apache Iceberg — el formato de tabla que absorbe buena parte de este patrón.
• Ejemplos companion — alineación de esquemas y dedup en capa de servicio ejecutables en Python y JavaScript con tests.