← volver

KStream en Kafka: lógica de negocio en movimiento

kafkakstreamstream-processing

¿qué es kstream?

kstream (kafka streams API) permite escribir lógica de negocio que consume, transforma y publica eventos en tiempo real, sin levantar clústeres externos.

se compila como una aplicación normal (java) y corre embebida; escala horizontalmente lanzando más instancias. ideal para pipelines livianos y de baja latencia.

KStream: el superpoder oculto de Kafka
idea — lógica de negocio sobre el stream.
idea clave: con kstream escribes transformaciones (filter, map, branch, join, aggregate) que corren de forma distribuida sobre tópicos de kafka. no requiere base de datos para topic→topic y puede mantener estado local (rocksdb) cuando agregas.

vista de alto nivel

una app lee de un tópico, aplica transformaciones y escribe en otro.

operaciones básicas

  • filter / mapValues: limpieza y enriquecimiento.
  • branch: split por condiciones.
  • to: publicación a tópico de salida.
flujo — filtro, mapeo, ramificación y publicación.

agregación con estado (state stores)

para conteos/agrupaciones se usan state stores locales (rocksdb) replicadas vía changelog. habilita groupByKey + aggregate/reduce con tolerancia a fallos.

agregados consistentes con recovery automático.

ejemplo real: “clean purchases”

el diagrama mapea las clases del ejemplo a cada bloque de la topología (main → topology → serde → modelo).

KStream: Clases y cómo se conectan en la topología
cajas: KafkaStreamsApplication, CleanPurchaseTopology, PurchaseSerde, Purchase.

cajas → clases

las cajas corresponden a las siguientes clases: KafkaStreamsApplication, CleanPurchaseTopology, PurchaseSerde y Purchase.

cleanpurchasetopology.java
package com.example.kstream.topologies;

import com.example.kstream.models.Purchase;
import com.example.kstream.serdes.PurchaseSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Kafka Streams topology for processing purchase events.
 * Filters valid purchases and transforms them to clean JSON.
 */
public class CleanPurchaseTopology {

    private static final Logger logger = LoggerFactory.getLogger(CleanPurchaseTopology.class);

    public Topology buildTopology(StreamsBuilder builder, String inputTopic, String outputTopic, Properties props) {

        builder.stream(inputTopic, Consumed.with(Serdes.String(), new PurchaseSerde()))
               .filter((key, event) -> {
                   boolean isValid = event != null && "OK".equalsIgnoreCase(event.getStatus());
                   if (!isValid) {
                       logger.info("Ignoring purchase with status: {}", event != null ? event.getStatus() : "null");
                   }
                   return isValid;
               })
               .mapValues(Purchase::toCleanJson)
               .to(outputTopic, Produced.with(Serdes.String(), Serdes.String()));

        return builder.build();
    }
}

topología mínima: filtra por status=ok, serializa y publica.

purchaseserde.java
package com.example.kstream.serdes;

import com.example.kstream.models.Purchase;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * Serde for Purchase class using Jackson.
 */
public class PurchaseSerde implements Serde<Purchase> {

    private final ObjectMapper mapper = new ObjectMapper();
    private static final Logger logger = LoggerFactory.getLogger(PurchaseSerde.class);

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {}

    @Override
    public Serializer<Purchase> serializer() {
        return (topic, data) -> {
            try {
                return mapper.writeValueAsBytes(data);
            } catch (Exception e) {
                logger.warn("Error serializing purchase: {}", e.getMessage());
                return null;
            }
        };
    }

    @Override
    public Deserializer<Purchase> deserializer() {
        return (topic, data) -> {
            try {
                return mapper.readValue(data, Purchase.class);
            } catch (Exception e) {
                logger.warn("Error deserializing purchase: {}", e.getMessage());
                return null;
            }
        };
    }
}

serializador/deserializador de la clase purchase. toma la clase purchase.java y la serializa para entrar al topico.

kafkastreamsapplication.java
package com.example.kstream;

import com.example.kstream.topologies.CleanPurchaseTopology;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;

/**
 * Entry point for the Kafka Streams application.
 * Sets up topology and configuration.
 */
public class KafkaStreamsApplication {

    private static final Logger logger = LoggerFactory.getLogger(KafkaStreamsApplication.class);

    public static void main(String[] args) throws Exception {

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "purchase-filter-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, "org.apache.kafka.common.serialization.Serdes$StringSerde");
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, "com.example.kstream.serdes.PurchaseSerde");
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        String inputTopic = "compras";
        String outputTopic = "compras-filtradas";

        StreamsBuilder builder = new StreamsBuilder();
        CleanPurchaseTopology topology = new CleanPurchaseTopology();
        KafkaStreams streams = new KafkaStreams(
                topology.buildTopology(builder, inputTopic, outputTopic, props),
                props
        );

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
        logger.info("Kafka Streams application started");
        runKafkaStreams(streams);
    }

    public static void runKafkaStreams(final KafkaStreams streams) throws InterruptedException {
        final CountDownLatch latch = new CountDownLatch(1);

        streams.setStateListener((newState, oldState) -> {
            if (oldState == KafkaStreams.State.RUNNING && newState != KafkaStreams.State.RUNNING) {
                latch.countDown();
            }
        });

        streams.start();

        try {
            logger.info("Kafka Streams state: {}", streams.state());
            latch.await();
        } catch (final InterruptedException e) {
            logger.error("Stream interrupted: {}", e.getMessage());
            Thread.currentThread().interrupt();
            throw e;
        }

        logger.info("Kafka Streams application closed");
    }
}

clase main donde se llama la topologia y se levanta el pod.

purchase.java
package com.example.kstream.models;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Modelo de datos para representar un evento de compra.
 */
@JsonIgnoreProperties(ignoreUnknown = true)
public class Purchase {
    private String id;
    private String status;
    private double amount;

    public String getId() {
        return id;
    }

    public String getStatus() {
        return status;
    }

    public double getAmount() {
        return amount;
    }

    public void setId(String id) {
        this.id = id;
    }

    public void setStatus(String status) {
        this.status = status;
    }

    public void setAmount(double amount) {
        this.amount = amount;
    }

    public String toCleanJson() {
        try {
            return new ObjectMapper().writeValueAsString(this);
        } catch (Exception e) {
            return "{}";
        }
    }
}

modelo de datos de compra.

beneficios

  • sin base de datos para transformaciones de paso.
  • escalable: replicas de la app = más throughput.
  • tiempo real: latencia ms-level; exactly-once opcional.
Sin base de datos, escalable, tiempo real
KStream = lógica de negocio en movimiento

recursos adicionales

🎓 Curso oficial y gratuito de Confluent sobre Kafka Streams (KStream y KTable):

Ir al curso de Kafka Streams en Confluent →