Kafka Stream Processing

Introducción

Apache Kafka es una plataforma distribuida de event streaming. Maneja trillones de eventos por día en empresas como LinkedIn, Uber y Netflix. A continuación: conceptos core de Kafka, producers, consumers, Kafka Streams API, Kafka Connect, Schema Registry y patrones comunes de stream processing con ejemplos en Python, Java y JavaScript.

Conceptos Core

• Topic: Un stream nombrado de eventos (como una categoría)
• Partition: Un topic se divide en partitions para paralelismo; cada partition es un log ordenado y append-only
• Offset: La posición de un mensaje dentro de una partition
• Consumer Group: Un grupo de consumers que comparten partitions de un topic
• Broker: Un servidor Kafka que almacena y sirve mensajes
• Producer: Una aplicación que publica eventos a topics
• Consumer: Una aplicación que se suscribe a topics y procesa eventos

Producer

Python (aiokafka)

from aiokafka import AIOKafkaProducer
import json
import asyncio

async def produce_events():
    producer = AIOKafkaProducer(
        bootstrap_servers="localhost:9092",
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
        key_serializer=lambda k: k.encode("utf-8") if k else None,
        acks="all",
        enable_idempotence=True,
    )
    await producer.start()
    try:
        for i in range(100):
            await producer.send_and_wait(
                "orders",
                key=f"order-{i}",
                value={"order_id": i, "amount": 99.99 * i, "status": "created"},
            )
    finally:
        await producer.stop()

asyncio.run(produce_events())

Java (Kafka Producer)

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class OrderProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all");
        props.put("enable.idempotence", "true");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(
                    "orders",
                    "order-" + i,
                    "{\"order_id\":" + i + ",\"amount\":" + (99.99 * i) + "}"
                );
                producer.send(record, (metadata, e) -> {
                    if (e != null) {
                        e.printStackTrace();
                    } else {
                        System.out.printf("Sent to partition %d offset %d%n",
                            metadata.partition(), metadata.offset());
                    }
                });
            }
        }
    }
}

Consumer

Python (aiokafka)

from aiokafka import AIOKafkaConsumer
import json
import asyncio

async def consume_events():
    consumer = AIOKafkaConsumer(
        "orders",
        bootstrap_servers="localhost:9092",
        group_id="order-processor",
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        key_deserializer=lambda k: k.decode("utf-8") if k else None,
        auto_offset_reset="earliest",
        enable_auto_commit=False,
    )
    await consumer.start()
    try:
        async for msg in consumer:
            print(f"Topic={msg.topic}, Partition={msg.partition}, Offset={msg.offset}")
            print(f"Key={msg.key}, Value={msg.value}")
            await consumer.commit()
    finally:
        await consumer.stop()

asyncio.run(consume_events())

JavaScript (KafkaJS)

const { Kafka } = require("kafkajs");

const kafka = new Kafka({
    clientId: "order-consumer",
    brokers: ["localhost:9092"],
});

const consumer = kafka.consumer({ groupId: "order-processor" });

async function consumeEvents() {
    await consumer.connect();
    await consumer.subscribe({ topic: "orders", fromBeginning: true });

    await consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
            const event = JSON.parse(message.value.toString());
            console.log(`Processing order ${event.order_id}, amount: ${event.amount}`);

            // Procesar la orden...
        },
    });
}

consumeEvents();

Kafka Streams (Java)

Kafka Streams es una librería para construir aplicaciones de stream processing en tiempo real.

Filter y branch

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;

public class OrderStreamProcessor {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("application.id", "order-processor");
        props.put("bootstrap.servers", "localhost:9092");
        props.put("default.key.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde");
        props.put("default.value.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> orders = builder.stream("orders");

        // Branch por amount
        KStream<String, String>[] branches = orders.branch(
            (key, value) -> value.contains("\"amount\":0"),
            (key, value) -> true
        );

        KStream<String, String> zeroOrders = branches[0];
        KStream<String, String> validOrders = branches[1];

        // Filter órdenes high-value
        KStream<String, String> highValue = validOrders.filter(
            (key, value) -> {
                // Parsear y chequear amount > 1000
                return value.contains("\"amount\":1");
            }
        );

        highValue.to("high-value-orders");
        validOrders.to("processed-orders");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

Aggregation (count por key)

KTable<String, Long> orderCounts = orders
    .groupBy((key, value) -> key, Grouped.with(
        Serdes.String(),
        Serdes.String()
    ))
    .count();

orderCounts.toStream().to("order-counts");

Windowed aggregation (ventanas tumbling de 5 minutos)

KTable<Windowed<String>, Long> windowedCounts = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count();

windowedCounts.toStream((key, value) -> key.key() + "@" + key.window().start())
    .to("windowed-order-counts");

Joining streams

KStream<String, String> orders = builder.stream("orders");
KStream<String, String> payments = builder.stream("payments");

KStream<String, String> enriched = orders.join(
    payments,
    (orderValue, paymentValue) -> orderValue + "|" + paymentValue,
    JoinWindows.of(Duration.ofMinutes(5)),
    Joined.with(Serdes.String(), Serdes.String(), Serdes.String())
);

enriched.to("enriched-orders");

Schema Registry

Schema Registry asegura compatibilidad entre producers y consumers enforceando un schema (Avro, Protobuf, JSON Schema).

Registrar un schema Avro

from confluent_kafka import SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

schema_registry_conf = {"url": "http://localhost:8081"}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

schema_str = """
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "int"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": "string"}
  ]
}
"""

avro_serializer = AvroSerializer(schema_registry_client, schema_str)

producer_conf = {
    "bootstrap.servers": "localhost:9092",
    "key.serializer": "org.apache.kafka.common.serialization.StringSerializer",
    "value.serializer": avro_serializer,
}

producer = SerializingProducer(producer_conf)
producer.produce(
    topic="orders-avro",
    key="order-1",
    value={"order_id": 1, "amount": 99.99, "status": "created"},
)
producer.flush()

Kafka Connect

Kafka Connect es un framework para conectar Kafka con sistemas externos (bases de datos, colas, file systems).

JDBC Source Connector

{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "postgres",
    "connection.password": "secret",
    "table.whitelist": "orders",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

Elasticsearch Sink Connector

{
  "name": "es-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "processed-orders",
    "connection.url": "http://localhost:9200",
    "type.name": "_doc",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "schema.ignore": "true"
  }
}

Patrones de Stream Processing

Patrón	Descripción	Caso de Uso
Filter	Descartar eventos que no matchean un predicado	Remover órdenes inválidas
Map	Transformar cada evento	Enriquecer con data de customer
Aggregate	Group y reduce	Contar órdenes por customer
Window	Agregar sobre tiempo	Rolling average de 5 minutos
Join	Combinar dos streams	Órdenes + pagos
Split	Rutear a múltiples topics	Órdenes high/low value
Repartition	Re-keyear un stream	Group por customer_id en vez de order_id

Pautas

• Usar acks=all para producers — asegura que los datos se escriban a múltiples réplicas
• Habilitar producers idempotentes — previene duplicados en retries
• Setear enable.auto.commit=false — commitear offsets después de procesar, no antes
• Manejar errores de deserialización — usar un dead-letter topic para poison pills
• Usar Schema Registry — enforce compatibilidad de schema, prevenir breaking consumers
• Particionar por key para ordering — eventos con el mismo key van a la misma partition
• Sizear partitions correctamente — apuntar a 10-50 MB/s throughput por partition
• Monitorear consumer lag — lag creciente significa que los consumers no pueden seguir
• Usar exactly-once semantics para pipelines críticos — setear processing.guarantee=exactly_once_v2
• Setear retención sabiamente — usar retención por tiempo (7 días) o por tamaño (1GB) por topic
• Usar consumer groups para paralelismo — cada consumer en un group obtiene un subset de partitions

Errores Comunes

• No manejar poison pills — un mensaje malo bloquea toda la partition
• Usar auto.offset.reset=latest — consumers nuevos se pierden datos históricos
• No setear max.poll.interval.ms — consumers lentos son kickeados del group
• Over-partitioning — demasiadas partitions aumentan overhead y latencia
• No monitorear lag — los issues aparecen solo cuando los usuarios se quejan
• Usar Kafka como database — Kafka es un log, no un store queryable
• No usar Schema Registry — cambios de schema rompen consumers silenciosamente
• Producir sin key — los eventos se distribuyen random, perdiendo garantías de ordering

See Also

• Stream Processing: Event-Driven Data Pipelines with
• Real-Time Analytics: From Events to Dashboards in Seconds
• Complete Guide to Event-Driven Systems
• Complete Guide to RabbitMQ Architecture
• Idempotent Consumer Pattern