Pipelines de Datos: Batch, Streaming, Lambda y Kappa

Introducción

El error de arquitectura más caro en ingeniería de datos no es elegir la herramienta equivocada: es montar infraestructura de streaming cuando un job batch nocturno habría sido suficiente. Los equipos adoptan Kafka y Flink porque el caso de uso suena a tiempo real, y después pasan meses manteniendo semántica de event-time, watermarks y lógica de replay para dashboards que los ejecutivos miran una vez al día.

Esta guía recorre las cuatro arquitecturas con las que te vas a encontrar — batch, streaming, Lambda y Kappa — más la decisión ETL vs ELT, que determina dónde corren las transformaciones. Cada sección incluye código funcional, y la tabla de decisión del final mapea requisitos con el patrón que encaja.

Resumen de los Patrones de Pipelines

Patrón	Latencia	Bases de código	Complejidad	Ideal para
Batch	Horas–días	1	Baja	Reportes, análisis histórico, joins pesados
Streaming	Segundos–minutos	1	Alta	Alertas, dashboards en vivo, sistemas orientados a eventos
Lambda	Ambas	2	Muy alta	Historia exacta + tiempo real, datos regulados
Kappa	Segundos–minutos	1	Alta	Equipos streaming-first con logs reproducibles
ETL	n/a	—	—	Transformar fuera del warehouse, enmascarar antes de cargar
ELT	n/a	—	—	Transformar dentro del warehouse con dbt/SQL

Los pipelines batch extraen, transforman y cargan según un calendario: Airflow, Spark, dbt y Snowflake son los sospechosos habituales. Los pipelines de streaming procesan los eventos a medida que llegan, a través de Kafka, Flink, Kinesis o Pulsar. Lambda mantiene una capa batch para la precisión histórica junto a una capa de velocidad para vistas en tiempo real — a costa de mantener la misma lógica dos veces. Kappa colapsa eso en una sola base de código de streaming, tratando el batch como el replay de un flujo acotado, lo que solo funciona si tu log retiene suficiente historia. ETL transforma antes de cargar; ELT carga primero los datos crudos y transforma dentro del warehouse.

Procesamiento por Lotes

Batch es donde la mayoría de los pipelines deberían empezar — y donde muchos deberían quedarse. Es más simple de depurar, más barato de correr, y una ejecución fallida simplemente se vuelve a correr.

Implementación con Airflow + Spark

Para un recorrido más profundo por Airflow (operators, XCom, sensors), mirá la guía de Apache Airflow.

# dags/etl_pipeline.py — DAG de Airflow para ETL batch
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "daily_etl_pipeline",
    default_args=default_args,
    schedule="0 2 * * *",  # Todos los días a las 2 AM
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
)

# Paso 1: extraer de las bases de datos de origen hacia S3
extract_orders = SparkSubmitOperator(
    task_id="extract_orders",
    conn_id="spark_default",
    application="/opt/airflow/jobs/extract_orders.py",
    application_args=["--date", "{{ ds }}", "--output", "s3://staging/orders/"],
    dag=dag,
)

# Paso 2: transformar con Spark
transform_orders = SparkSubmitOperator(
    task_id="transform_orders",
    conn_id="spark_default",
    application="/opt/airflow/jobs/transform_orders.py",
    application_args=[
        "--input", "s3://staging/orders/{{ ds }}/",
        "--output", "s3://processed/orders/{{ ds }}/",
    ],
    dag=dag,
)

# Paso 3: cargar en Snowflake
load_orders = SnowflakeOperator(
    task_id="load_orders",
    sql="""
        COPY INTO orders_staging
        FROM s3://processed/orders/{{ ds }}/
        STORAGE_INTEGRATION = orders_s3_int
        FILE_FORMAT = (TYPE = PARQUET);
    """,
    snowflake_conn_id="snowflake_default",
    dag=dag,
)

# Paso 4: modelos de dbt
run_dbt_models = BashOperator(
    task_id="run_dbt_models",
    bash_command="cd /opt/dbt && dbt run --select staging+ --vars '{date: {{ ds }}}'",
    dag=dag,
)

extract_orders >> transform_orders >> load_orders >> run_dbt_models

Fijate en el STORAGE_INTEGRATION del COPY INTO: referencia una integración de almacenamiento de Snowflake respaldada por un rol de IAM. Los ejemplos más viejos interpolan claves de AWS dentro del SQL — esas terminan en el historial de queries, en los logs y en los templates renderizados de Airflow, así que no copies ese patrón.

Job batch de Spark

# jobs/transform_orders.py — transformación batch con Spark
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import argparse

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    spark = SparkSession.builder.appName("TransformOrders").getOrCreate()

    # Leer desde staging
    df = spark.read.parquet(args.input)

    # Transformar: limpiar, enriquecer, agregar
    transformed = (
        df
        .filter(F.col("status").isin("completed", "shipped"))
        .withColumn("order_date", F.to_date("created_at"))
        .withColumn("total_amount", F.col("subtotal") + F.col("tax") + F.col("shipping"))
        .groupBy("order_date", "region")
        .agg(
            F.count("*").alias("order_count"),
            F.sum("total_amount").alias("revenue"),
            F.avg("total_amount").alias("avg_order_value"),
        )
    )

    # Escribir a la zona procesada — overwrite hace la corrida idempotente
    transformed.write.mode("overwrite").partitionBy("order_date").parquet(args.output)
    spark.stop()

if __name__ == "__main__":
    main()

Procesamiento en Streaming

El streaming justifica su complejidad cuando el negocio realmente reacciona a los eventos en segundos: scoring de fraude, inventario en vivo, alertas en tiempo real. Si el consumidor lee los resultados a la mañana siguiente, con batch alcanzaba.

Para el deep-dive completo de streaming — manejo de estado, semántica exactly-once, Kafka Streams vs Flink — mirá la guía de stream processing.

Implementación con Kafka + Flink

// OrderStreamingJob.java — job de streaming para Flink 1.17+/2.x
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.formats.json.JsonNodeDeserializationSchema;
import org.apache.flink.formats.json.JsonNodeSerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.time.Duration;

public class OrderStreamingJob {

    // AggregateFunction: el acumulador es {orderCount, totalRevenue}
    public static class OrderAggregator
            implements AggregateFunction<JsonNode, double[], JsonNode> {

        @Override public double[] createAccumulator() { return new double[]{0, 0.0}; }

        @Override public double[] add(JsonNode order, double[] acc) {
            acc[0] += 1;
            acc[1] += order.get("total_amount").asDouble();
            return acc;
        }

        @Override public JsonNode getResult(double[] acc) {
            ObjectNode out = new ObjectMapper().createObjectNode();
            out.put("order_count", (int) acc[0]);
            out.put("total_revenue", acc[1]);
            return out;
        }

        @Override public double[] merge(double[] a, double[] b) {
            a[0] += b[0]; a[1] += b[1];
            return a;
        }
    }

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

        // Consumir desde Kafka
        KafkaSource<JsonNode> source = KafkaSource.<JsonNode>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("orders")
            .setGroupId("order-processor")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new JsonNodeDeserializationSchema())
            .build();

        DataStream<JsonNode> orders = env.fromSource(
            source,
            // Las ventanas de event-time necesitan watermarks o nunca disparan
            WatermarkStrategy.<JsonNode>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                .withTimestampAssigner((order, ts) -> order.get("created_at").asLong()),
            "Kafka Orders"
        );

        // Procesar: filtrar, enriquecer, agregar
        DataStream<JsonNode> aggregated = orders
            .filter(node -> node.has("status") && "completed".equals(node.get("status").asText()))
            .map(node -> {
                ObjectNode enriched = node.deepCopy();
                double total = node.get("subtotal").asDouble()
                    + node.get("tax").asDouble()
                    + node.get("shipping").asDouble();
                enriched.put("total_amount", total);
                return (JsonNode) enriched;
            })
            .returns(JsonNode.class)
            // Ventanas tumbling de 5 minutos sobre event-time
            .keyBy(node -> node.get("region").asText())
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new OrderAggregator());

        // Escribir al topic de salida de Kafka
        KafkaSink<JsonNode> sink = KafkaSink.<JsonNode>builder()
            .setBootstrapServers("kafka:9092")
            .setRecordSerializer(
                KafkaRecordSerializationSchema.builder()
                    .setTopic("order-metrics")
                    .setValueSerializationSchema(new JsonNodeSerializationSchema())
                    .build()
            )
            .build();

        aggregated.sinkTo(sink);
        env.execute("Order Streaming Job");
    }
}

Hay dos detalles que importan acá y es fácil equivocarse. FlinkKafkaConsumer/FlinkKafkaProducer quedaron deprecados en Flink 1.17 y se eliminaron en Flink 2.x — usá KafkaSource/KafkaSink. Y las ventanas de event-time sin un WatermarkStrategy nunca disparan, que es el clásico bug de “mi job corre pero no produce nada”.

Streaming en Python con Faust

# streaming/order_processor.py — pip install faust-streaming (fork mantenido)
import faust

app = faust.App(
    "order-processor",
    broker="kafka://kafka:9092",
    store="rocksdb://",
)

class Order(faust.Record, serializer="json"):
    order_id: str
    customer_id: str
    status: str
    subtotal: float
    tax: float
    shipping: float
    region: str

class OrderMetric(faust.Record, serializer="json"):
    region: str
    order_count: int
    total_revenue: float

orders_topic = app.topic("orders", value_type=Order)
metrics_topic = app.topic("order-metrics", value_type=OrderMetric)

# Tablas con estado — tienen que existir antes del agent que las puebla
region_counts = app.Table("region_counts", default=int)
region_revenue = app.Table("region_revenue", default=float)

@app.agent(orders_topic)
async def process_order(orders):
    async for order in orders:
        if order.status == "completed":
            total = order.subtotal + order.tax + order.shipping
            region_counts[order.region] += 1
            region_revenue[order.region] += total

# Cada 5 minutos, emitir la ventana acumulada y reiniciarla
@app.timer(interval=300.0)
async def emit_windowed_metrics():
    for region in list(region_counts.keys()):
        await metrics_topic.send(value=OrderMetric(
            region=region,
            order_count=region_counts[region],
            total_revenue=region_revenue[region],
        ))
        region_counts[region] = 0
        region_revenue[region] = 0.0

if __name__ == "__main__":
    app.main()

Una advertencia: faust-streaming es el fork mantenido del paquete faust original, abandonado, pero su propio ritmo de releases es lento. Para proyectos nuevos, evaluá Bytewax, Quix Streams o consumidores simples con kafka-python antes de comprometerte.

Arquitectura Lambda

Lambda corre una capa batch para la historia exacta y una capa de velocidad para aproximaciones en tiempo real, combinando ambas al momento de la consulta. Fue diseñada cuando el procesamiento de streams no podía garantizar corrección — hoy esa garantía existe, y por eso la mayoría de los sistemas nuevos la saltean.

• Ventajas: precisión histórica y latencia de tiempo real a la vez; tolerante a datos tardíos o reprocesados.
• Desventajas: cada regla de negocio existe dos veces, en dos frameworks — el modo de fallo clásico de Lambda es que las dos capas divergen y la capa de servicio devuelve respuestas distintas sin que nadie lo note.

Arquitectura Kappa

Kappa es Lambda sin la capa batch: una sola base de código de streaming, y “batch” es reproducir el log desde un offset anterior. El backfilling es un job de reprocesamiento, no un pipeline aparte.

• Ventajas: una base de código, un solo modelo de procesamiento; el replay te da backfilling gratis.
• Desventajas: requiere un log con capacidad de replay, así que ventanas de reprocesamiento largas implican mucho almacenamiento en Kafka; si tu job “batch” es un join de Spark de 6 horas, reproducirlo a través de un procesador de streams duele.

ETL vs ELT

Este es un eje separado del de batch vs streaming: se trata de dónde ocurre la T.

	ETL	ELT
Orden	Extraer → Transformar → Cargar	Extraer → Cargar → Transformar
Dónde corre la T	Fuera del warehouse (Spark, Python)	Dentro del warehouse (dbt, SQL)
El warehouse guarda	Solo datos limpios y modelados	Crudos + transformados (staging → marts)
Elegir cuando	El cómputo del warehouse es limitado; hay que enmascarar PII antes de que aterrice	El warehouse es potente (Snowflake, BigQuery); querés desarrollo iterativo con dbt
Stack típico	Airflow + Spark, Informatica, Talend	Fivetran + dbt, Snowpipe + dbt

La guía de pipelines ETL cubre cargas incrementales, manejo de SCD y patrones idempotentes en detalle.

ELT con dbt

En dbt cada modelo es su propio archivo .sql — los dos de abajo se muestran juntos por legibilidad.

-- models/staging/stg_orders.sql — Modelo de staging (crudo → limpio)
SELECT
    order_id,
    customer_id,
    CAST(created_at AS TIMESTAMP) AS created_at,
    CAST(subtotal AS DECIMAL(10,2)) AS subtotal,
    CAST(tax AS DECIMAL(10,2)) AS tax,
    CAST(shipping AS DECIMAL(10,2)) AS shipping,
    status,
    region
FROM {{ source('raw', 'orders') }}
WHERE status IS NOT NULL

-- models/marts/fct_daily_revenue.sql — Modelo de mart (limpio → negocio)
SELECT
    DATE(created_at) AS order_date,
    region,
    COUNT(*) AS order_count,
    SUM(subtotal + tax + shipping) AS total_revenue,
    AVG(subtotal + tax + shipping) AS avg_order_value
FROM {{ ref('stg_orders') }}
WHERE status = 'completed'
GROUP BY 1, 2

Cómo Elegir el Patrón Correcto

Requisito	Patrón
Reportes diarios/semanales, mucha historia, joins pesados	Batch
Frescura medida en horas; sensible al costo	Batch
Alertas y dashboards consumidos en segundos	Streaming
Sistemas orientados a eventos que reaccionan a cada evento	Streaming
Historia exacta y tiempo real; necesidades de auditoría/regulatorias	Lambda
Una sola base de código; la retención del log cubre tu ventana de backfill	Kappa
Enmascarar PII antes de que los datos aterricen en el warehouse	ETL
Warehouse potente; equipo SQL-native; modelado iterativo	ELT

Y cuándo no ir por la opción sofisticada: si nadie consulta los datos más rápido que una vez al día, el streaming es puro overhead — complejidad operativa, guardias y costo de infraestructura a cambio de una frescura que nadie consume. La misma lógica aplica a Lambda: mantener dos bases de código solo tiene sentido cuando ambas vistas realmente se consultan.

Para una implementación completa del patrón Lambda — ingesta con Kafka Connect, consultas de merge con Trino, deduplicación — mirá el patrón puente batch-streaming.

Juntando Todo: Una Arquitectura Real

En la práctica estos patrones se componen en vez de competir. Una plataforma de datos típica de e-commerce se ve así:

1. El servicio de órdenes escribe en Postgres. Un conector de CDC (Debezium) lleva los cambios a Kafka — ese es tu flujo de eventos crudos.
2. Rama de streaming: un job de Flink lee el topic orders y calcula señales de fraude y conteos de inventario en vivo en segundos. Solo viven acá los consumidores que realmente necesitan latencia menor al minuto.
3. Rama batch: el mismo topic de Kafka aterriza cada hora en S3 como Parquet. Un DAG nocturno de Airflow (el de arriba) corre los modelos de dbt que producen fct_daily_revenue y el resto de los marts del warehouse — la mitad ELT de la imagen.
4. Servicio: los dashboards leen del warehouse; la API de fraude lee del destino de streaming. Nadie mantiene dos versiones de la lógica de revenue, porque no existe un duplicado de ella en una capa de velocidad — eso sería Lambda, y es la parte que deliberadamente evitás.

El flujo de decisión es: identificar el requisito de frescura de cada consumidor, pasar a streaming solo las ramas que lo necesitan y dejar todo lo demás en el camino batch, aburrido pero confiable. Cuando un consumidor más adelante necesite datos más rápidos, ya tenés el log con replay — esa es la válvula de escape estilo Kappa para el backfill.

Mejores Prácticas

• Empezar con batch y agregar streaming solo cuando un consumidor real necesite la latencia — batch es más simple y más barato.
• Ir por ELT por defecto cuando el warehouse lo soporta; dbt + Snowflake/BigQuery es donde la mayoría de los equipos termina hoy.
• Retener los logs de Kafka el tiempo suficiente para reproducirlos — esa ventana de retención es lo que hace posible Kappa y el backfilling.
• Hacer cada pipeline idempotente: volver a correrlo tiene que producir el mismo resultado, con merge/upsert y particionado por fecha.
• Monitorear la calidad de los datos, no solo el estado del job — conteos de filas, tasas de nulos, drift de esquema y frescura. Mirá la guía de calidad de datos.
• Registrar los esquemas de cada topic de Kafka para que productores y consumidores no diverjan en silencio.
• Activar checkpoints en los jobs de streaming — los checkpoints de Flink/Spark son lo que hace real el exactly-once.
• Mantener las zonas separadas: raw → staging → marts.
• Alertar sobre fallos de pipeline; un retry silencioso de Airflow que esconde una fuente rota es la forma en que los datos malos llegan a producción.
• Documentar el linaje con dbt docs u OpenLineage antes de que alguien pregunte de dónde salió un número.

Errores Comunes

• Empezar con streaming: es la forma más cara de descubrir que con batch alcanzaba. Ganate la complejidad.
• Sin idempotencia: volver a correr un pipeline agrega datos duplicados. Usar patrones de merge/upsert.
• Sin gestión de esquemas: productores y consumidores divergen. Usar un registro de esquemas.
• Quedarse con Lambda demasiado tiempo: las dos bases de código divergen con el tiempo y la capa de velocidad se convierte en la respuesta equivocada-pero-rápida. Si ya la tenés, planificá la migración a Kappa; si no, no empieces una.
• Sin chequeos de calidad de datos: los pipelines producen datos erróneos en silencio. Agregar Great Expectations o tests de dbt.

Lecturas Recomendadas

• Documentación de Apache Airflow — DAGs, operators y scheduling
• Documentación de Apache Flink — KafkaSource/KafkaSink, watermarks, checkpoints
• Documentación de dbt — sources, refs y estructura de proyectos
• Questioning the Lambda Architecture — el argumento original de Jay Kreps a favor de Kappa
• Documentación de Apache Kafka — retención del log y semántica de replay