🚀 Ingestión de Datos en Tiempo Real: Construyendo un Pipeline con Apache Kafka y Flink

En el mundo actual, la capacidad de procesar y analizar datos a medida que se generan es crucial para obtener insights rápidos y tomar decisiones oportunas. La ingestión de datos en tiempo real es la piedra angular de esta capacidad, permitiendo a las organizaciones reaccionar instantáneamente a eventos críticos.

Este tutorial te sumergirá en el fascinante universo de la ingestión de datos en tiempo real, mostrándote cómo construir un pipeline eficiente y escalable utilizando dos de las herramientas más potentes del ecosistema Big Data: Apache Kafka para la mensajería de datos y Apache Flink para el procesamiento de streams.

🎯 ¿Por qué Ingestión de Datos en Tiempo Real?

Imagina un escenario donde necesitas detectar fraudes financieros al instante, monitorear la salud de sensores IoT en milisegundos o personalizar la experiencia de usuario en una aplicación de comercio electrónico en tiempo real. En todos estos casos, el procesamiento por lotes tradicional (batch processing) simplemente no es suficiente. Necesitamos la velocidad y agilidad que solo los sistemas en tiempo real pueden ofrecer.

📈 Ventajas Clave del Procesamiento en Tiempo Real:

• Decisiones Oportunas: Reacciona a eventos críticos casi al instante.
• Experiencia de Usuario Mejorada: Personalización y respuestas dinámicas en aplicaciones.
• Detección de Anomalías: Identifica patrones inusuales o fraudes en el momento que ocurren.
• Monitoreo Continuo: Supervisión constante de sistemas y métricas.
• Reducción de Latencia: Minimiza el retraso entre la generación del dato y su disponibilidad para el análisis.

🛠️ Herramientas Fundamentales: Kafka y Flink

Para construir nuestro pipeline, nos centraremos en dos gigantes de la infraestructura de datos en tiempo real:

📨 Apache Kafka: El Sistema de Mensajería Distribuido

Apache Kafka es una plataforma de streaming distribuida que permite publicar, suscribirse, almacenar y procesar flujos de registros en tiempo real. Es el cerebro que transportará nuestros datos desde su origen hasta su destino de procesamiento.

Características Clave de Kafka:

• Alta Disponibilidad y Tolerancia a Fallos: Datos replicados en múltiples nodos.
• Escalabilidad Horizontal: Añade más brokers para aumentar la capacidad.
• Durabilidad: Los datos persisten en disco por un período configurable.
• Publicación/Suscripción: Patrón de mensajería asíncrona.
• Rendimiento: Diseñado para manejar altos volúmenes y tasas de transferencia.

⚡ Apache Flink: El Motor de Procesamiento de Streams

Apache Flink es un potente motor de procesamiento de streams y batch unificado. Su principal fortaleza reside en su capacidad para procesar datos en tiempo real con baja latencia, alta tasa de transferencia y semántica de estado consistente, incluso en caso de fallos.

Características Clave de Flink:

• Procesamiento de Streams Verdadero: Manejo de eventos uno por uno con baja latencia.
• Semántica de Estado Potente: Gestiona estados consistentes incluso con checkpoints y recuperación de fallos.
• Ventanas Flexibles: Soporte para ventanas de tiempo (tumbling, sliding, session) y ventanas de conteo.
• Tolerancia a Fallos: Mediante checkpoints asíncronos y recuperación de estados.
• Escalabilidad: Distribuido y capaz de procesar petabytes de datos.

🗺️ Arquitectura del Pipeline de Ingestión

Nuestro pipeline de ingestión de datos en tiempo real seguirá un patrón común:

1. Fuente de Datos (Productor): Genera eventos.
2. Kafka (Broker): Recibe y almacena eventos en tópicos.
3. Flink (Consumidor/Procesador): Lee eventos de Kafka, los procesa y transforma.
4. Sumidero (Sink): Almacena los datos procesados en una base de datos, data lake, o los envía a otra aplicación.

Veamos un diagrama visual de esta arquitectura:

Ejemplo de Flujo de Datos:

1. Un sensor IoT (Fuente de Datos) genera un evento de temperatura cada segundo.
2. El evento se envía a un tópico de Kafka llamado temperatura_raw.
3. Un trabajo de Flink lee de temperatura_raw, filtra lecturas anómalas y calcula la temperatura promedio cada 10 segundos.
4. Los datos procesados se escriben en una base de datos NoSQL (Sumidero de Datos) como Cassandra o Elasticsearch, o en un nuevo tópico de Kafka temperatura_procesada para otro consumidor.

⚙️ Preparación del Entorno

Antes de sumergirnos en el código, necesitamos configurar nuestro entorno. Para este tutorial, asumiremos que tienes Docker instalado, lo cual simplifica enormemente la gestión de Kafka y Flink.

🐳 Docker Compose: Un Orquestador Sencillo

Crearemos un archivo docker-compose.yml para levantar Kafka (con Zookeeper, su dependencia) y Flink.

Crea un directorio realtime-pipeline y dentro de él, el archivo docker-compose.yml:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  jobmanager:
    image: apache/flink:1.17.1-scala_2.12
    hostname: jobmanager
    container_name: jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |-
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
    depends_on:
      - kafka # Asegúrate de que Kafka esté listo antes que Flink

  taskmanager:
    image: apache/flink:1.17.1-scala_2.12
    hostname: taskmanager
    container_name: taskmanager
    command: taskmanager
    environment:
      - |-
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
      - TASK_MANAGER_NUMBER_OF_TASK_SLOTS=1
    depends_on:
      - jobmanager

Para levantar los servicios, ejecuta en tu terminal desde el directorio realtime-pipeline:

docker-compose up -d

Verifica que todo esté corriendo con:

docker-compose ps

Deberías ver los contenedores zookeeper, kafka, jobmanager y taskmanager en estado Up.

📝 Creando un Productor de Kafka (Python)

Ahora, escribiremos un script de Python simple que actúe como nuestra fuente de datos, generando mensajes y enviándolos a un tópico de Kafka. Este script simulará un sensor de temperatura.

Crea un archivo producer.py en el mismo directorio:

from kafka import KafkaProducer
import json
import time
import random

# Configuración del productor de Kafka
KAFKA_BROKER = 'localhost:9092' # O 'kafka:29092' si ejecutas desde otro contenedor Docker
KAFKA_TOPIC = 'temperatura_raw'

producer = KafkaProducer(
    bootstrap_servers=[KAFKA_BROKER],
    value_serializer=lambda v: json.dumps(v).encode('utf-8') # Serializa mensajes a JSON y luego a bytes
)

def generate_temperature_data():
    timestamp = int(time.time() * 1000) # Timestamp en milisegundos
    device_id = f"sensor_{random.randint(1, 10)}"
    temperature = round(random.uniform(20.0, 30.0), 2) # Temperatura entre 20 y 30 grados
    humidity = round(random.uniform(40.0, 60.0), 2) # Humedad entre 40 y 60%
    
    return {
        'timestamp': timestamp,
        'device_id': device_id,
        'temperature': temperature,
        'humidity': humidity
    }

print(f"Iniciando productor de Kafka para el tópico '{KAFKA_TOPIC}'...")
try:
    while True:
        data = generate_temperature_data()
        producer.send(KAFKA_TOPIC, value=data)
        print(f"Enviado: {data}")
        time.sleep(1) # Envía un mensaje cada segundo
except KeyboardInterrupt:
    print("\nProductor detenido.")
finally:
    producer.close()

Instala la librería kafka-python:

pip install kafka-python

Ejecuta el productor:

python producer.py

Verás mensajes de temperatura generándose y enviándose a Kafka. Podemos verificar que el tópico se ha creado y que los mensajes están llegando usando la CLI de Kafka (dentro del contenedor kafka):

docker exec -it kafka kafka-topics --bootstrap-server localhost:29092 --list
# Deberías ver 'temperatura_raw'

docker exec -it kafka kafka-console-consumer --bootstrap-server localhost:29092 --topic temperatura_raw --from-beginning
# Deberías ver los mensajes JSON que envía tu productor

✨ Procesando Streams con Apache Flink (Java/Scala)

Ahora viene la parte emocionante: procesar estos datos en tiempo real con Flink. Usaremos la API de DataStream de Flink para construir nuestro trabajo.

Crearemos un proyecto Maven simple. En el directorio realtime-pipeline, crea una subcarpeta flink-processor.

Dentro de flink-processor, crea un archivo pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>flink-processor</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>Flink Processor</name>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.17.1</flink.version>
        <java.version>11</java.version>
        <scala.binary.version>2.12</scala.binary.version>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
    </properties>

    <dependencies>
        <!-- Apache Flink dependencies -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        
        <!-- Flink Kafka Connector -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- JSON parsing -->
        <dependency>
            <groupId>org.json</groupId>
            <artifactId>json</artifactId>
            <version>20231013</version>
        </dependency>

        <!-- Logging -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.36</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <finalName>${project.artifactId}-${project.version}</finalName>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Crea el directorio src/main/java/com/example/flink y dentro de él, el archivo TemperatureProcessor.java:

package com.example.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
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 org.apache.flink.util.Collector;
import org.json.JSONObject;

import java.time.Duration;

public class TemperatureProcessor {

    public static void main(String[] args) throws Exception {
        // 1. Configurar el entorno de ejecución de Flink
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // Para este ejemplo simple, un paralelismo de 1 es suficiente
        env.enableCheckpointing(60000); // Habilitar checkpointing cada 60 segundos para tolerancia a fallos

        // 2. Configurar el origen de Kafka (Source)
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("kafka:29092") // El nombre del servicio Kafka en docker-compose
                .setTopics("temperatura_raw")
                .setGroupId("flink-temperature-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 3. Crear un DataStream a partir de Kafka
        DataStream<String> rawReadings = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // 4. Mapear y transformar los datos JSON
        DataStream<SensorReading> sensorReadings = rawReadings.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String value) throws Exception {
                JSONObject json = new JSONObject(value);
                long timestamp = json.getLong("timestamp");
                String deviceId = json.getString("device_id");
                double temperature = json.getDouble("temperature");
                double humidity = json.getDouble("humidity");
                return new SensorReading(timestamp, deviceId, temperature, humidity);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> event.getTimestamp()));

        // 5. Procesar los datos: Calcular la temperatura promedio por dispositivo en ventanas de 10 segundos
        DataStream<String> processedReadings = sensorReadings
                .keyBy(SensorReading::getDeviceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(new TemperatureAggregator())
                .map(new MapFunction<AggregatedTemperature, String>() {
                    @Override
                    public String map(AggregatedTemperature agg) throws Exception {
                        return String.format("[%s] Promedio de Temperatura para %s: %.2f (con %d lecturas)",
                                agg.getWindowEnd(), agg.getDeviceId(), agg.getAverageTemperature(), agg.getCount());
                    }
                });

        // 6. Imprimir los resultados procesados (Sumidero - Sink)
        processedReadings.print().name("Processed Temperatures");

        // 7. Ejecutar el trabajo de Flink
        env.execute("Realtime Temperature Processor");
    }

    // Clases de datos (POJOs) para representar las lecturas del sensor y los resultados agregados
    public static class SensorReading {
        private long timestamp;
        private String deviceId;
        private double temperature;
        private double humidity;

        public SensorReading() {}

        public SensorReading(long timestamp, String deviceId, double temperature, double humidity) {
            this.timestamp = timestamp;
            this.deviceId = deviceId;
            this.temperature = temperature;
            this.humidity = humidity;
        }

        public long getTimestamp() { return timestamp; }
        public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
        public String getDeviceId() { return deviceId; }
        public void setDeviceId(String deviceId) { this.deviceId = deviceId; }
        public double getTemperature() { return temperature; }
        public void setTemperature(double temperature) { this.temperature = temperature; }
        public double getHumidity() { return humidity; }
        public void setHumidity(double humidity) { this.humidity = humidity; }

        @Override
        public String toString() {
            return "SensorReading{" +
                   "timestamp=" + timestamp +
                   ", deviceId='" + deviceId + '\'' +
                   ", temperature=" + temperature +
                   ", humidity=" + humidity +
                   '}';
        }
    }

    public static class AggregatedTemperature {
        private String deviceId;
        private double totalTemperature;
        private long count;
        private long windowEnd;

        public AggregatedTemperature() {}

        public AggregatedTemperature(String deviceId, double totalTemperature, long count, long windowEnd) {
            this.deviceId = deviceId;
            this.totalTemperature = totalTemperature;
            this.count = count;
            this.windowEnd = windowEnd;
        }

        public String getDeviceId() { return deviceId; }
        public void setDeviceId(String deviceId) { this.deviceId = deviceId; }
        public double getTotalTemperature() { return totalTemperature; }
        public void setTotalTemperature(double totalTemperature) { this.totalTemperature = totalTemperature; }
        public long getCount() { return count; }
        public void setCount(long count) { this.count = count; }
        public long getWindowEnd() { return windowEnd; }
        public void setWindowEnd(long windowEnd) { this.windowEnd = windowEnd; }

        public double getAverageTemperature() {
            return count > 0 ? totalTemperature / count : 0.0;
        }
    }

    public static class TemperatureAggregator implements org.apache.flink.api.common.functions.AggregateFunction<SensorReading, AggregatedTemperature, AggregatedTemperature> {

        @Override
        public AggregatedTemperature createAccumulator() {
            return new AggregatedTemperature(null, 0.0, 0L, 0L);
        }

        @Override
        public AggregatedTemperature add(SensorReading value, AggregatedTemperature accumulator) {
            if (accumulator.getDeviceId() == null) {
                accumulator.setDeviceId(value.getDeviceId());
            }
            accumulator.setTotalTemperature(accumulator.getTotalTemperature() + value.getTemperature());
            accumulator.setCount(accumulator.getCount() + 1);
            return accumulator;
        }

        @Override
        public AggregatedTemperature getResult(AggregatedTemperature accumulator) {
            return accumulator;
        }

        @Override
        public AggregatedTemperature merge(AggregatedTemperature a, AggregatedTemperature b) {
            return new AggregatedTemperature(
                    a.getDeviceId(),
                    a.getTotalTemperature() + b.getTotalTemperature(),
                    a.getCount() + b.getCount(),
                    Math.max(a.getWindowEnd(), b.getWindowEnd()) // Esto podría ser más sofisticado para windowEnd real
            );
        }
    }
}

Explicación del Código Flink:

1. StreamExecutionEnvironment: El punto de entrada para construir un trabajo de Flink. Configuramos el paralelismo y el checkpointing para tolerancia a fallos.
2. KafkaSource: Define cómo Flink se conecta a Kafka, qué tópico leer (temperatura_raw), y cómo deserializar los mensajes (usamos SimpleStringSchema para obtener Strings).
3. WatermarkStrategy: Esencial para el procesamiento de eventos en tiempo. Define cómo Flink maneja el orden de los eventos y el procesamiento de datos fuera de orden (forBoundedOutOfOrderness es común).
4. map: Transforma la cadena JSON cruda en un objeto SensorReading de Java para facilitar el acceso a los campos.
5. keyBy: Agrupa los eventos por deviceId. Esto asegura que todas las lecturas de un mismo sensor vayan al mismo operador de agregación.
6. window(TumblingEventTimeWindows.of(Time.seconds(10))): Define una ventana de tiempo tumbling de 10 segundos. Esto significa que cada 10 segundos, se calculará el resultado para los eventos que cayeron en esa ventana.
7. aggregate(new TemperatureAggregator()): Aplica una función de agregación personalizada (TemperatureAggregator) dentro de cada ventana. Esta función calcula la suma de temperaturas y el conteo de lecturas.
8. print(): Un sumidero simple que imprime los resultados en la consola de Flink JobManager/TaskManager. En un entorno de producción, esto se reemplazaría por un sumidero a una base de datos, otro tópico de Kafka, etc.

Construcción y Ejecución del Trabajo Flink:

Navega al directorio flink-processor en tu terminal y compila el proyecto Maven para crear un fat JAR:

mvn clean package

Esto creará un archivo JAR ejecutable en target/flink-processor-1.0-SNAPSHOT.jar.

Ahora, envía este JAR al JobManager de Flink. Puedes hacerlo a través de la interfaz web de Flink (http://localhost:8081) subiendo el JAR, o usando la CLI de Flink (que ya está disponible dentro del contenedor jobmanager).

La forma más sencilla usando la CLI (desde el directorio realtime-pipeline):

docker cp flink-processor/target/flink-processor-1.0-SNAPSHOT.jar jobmanager:/opt/flink/usrlib/
docker exec -it jobmanager bash -c "flink run -c com.example.flink.TemperatureProcessor /opt/flink/usrlib/flink-processor-1.0-SNAPSHOT.jar"

Una vez ejecutado, verás en la consola de docker-compose logs -f jobmanager y docker-compose logs -f taskmanager la salida de Flink con los promedios de temperatura calculados cada 10 segundos por cada sensor. Si todo funciona correctamente, el JobManager mostrará la salida del print() de Flink.

📊 Resultados y Verificación

Si has seguido todos los pasos, tu pipeline está funcionando:

1. El producer.py está generando lecturas de sensores y enviándolas a Kafka.
2. El tópico temperatura_raw de Kafka está recibiendo y almacenando esos mensajes.
3. El trabajo de Flink está leyendo del tópico, transformando los datos y calculando promedios por dispositivo en ventanas de 10 segundos.
4. Los resultados agregados se están imprimiendo en los logs del JobManager/TaskManager.

Aquí tienes un ejemplo de cómo se vería la salida en los logs de Flink:

jobmanager    | 16:03:10,000 INFO  com.example.flink.TemperatureProcessor [] - 9> [1701465800000] Promedio de Temperatura para sensor_5: 25.50 (con 10 lecturas)
jobmanager    | 16:03:10,000 INFO  com.example.flink.TemperatureProcessor [] - 9> [1701465800000] Promedio de Temperatura para sensor_2: 26.12 (con 10 lecturas)
jobmanager    | 16:03:10,000 INFO  com.example.flink.TemperatureProcessor [] - 9> [1701465800000] Promedio de Temperatura para sensor_8: 24.88 (con 10 lecturas)

Los timestamps de las ventanas serán múltiplos de 10 segundos, reflejando el final de cada ventana.

⏭️ Próximos Pasos y Consideraciones Avanzadas

Este tutorial te ha proporcionado una base sólida para la ingestión de datos en tiempo real. Aquí hay algunas áreas para explorar y mejorar tu pipeline:

✅ Monitoreo y Escalabilidad

• Prometheus/Grafana: Integra métricas de Kafka y Flink para una visibilidad completa del rendimiento del pipeline.
• Autoscaling: Configura tu clúster de Flink para escalar automáticamente los TaskManagers en función de la carga.
• Kafka Connect: Para conectar Kafka a diversas fuentes y sumideros de datos de forma declarativa sin escribir código.

🛡️ Tolerancia a Fallos y Consistencia

• Checkpointing de Flink: Ajusta la frecuencia de los checkpoints y su ubicación de almacenamiento (HDFS, S3) para una recuperación robusta ante fallos.
• Exactly-Once Semantics: Explora cómo Flink y Kafka pueden garantizar la semántica exactly-once para evitar la duplicación o pérdida de datos.

📈 Más allá de print()

• Sumideros (Sinks) Reales: Envía los datos procesados a bases de datos (Cassandra, Elasticsearch, InfluxDB), data lakes (S3, HDFS), o incluso a otros tópicos de Kafka para un procesamiento posterior.
  • Por ejemplo, podrías escribir en un nuevo tópico temperatura_promedio y tener otra aplicación consumiéndolo para visualización.

💡 Optimización del Rendimiento

• Configuración de Flink: Experimenta con el paralelismo, tamaño de memoria, y configuración de red para optimizar el rendimiento de tu trabajo.
• Serialización: Considera formatos de serialización más eficientes como Apache Avro o Protobuf en lugar de JSON para reducir el tamaño de los mensajes y mejorar el rendimiento.

Conclusion

¡Felicidades! Has construido un pipeline de ingestión de datos en tiempo real funcional utilizando Apache Kafka y Apache Flink. Has aprendido los fundamentos de cómo estas tecnologías trabajan juntas para capturar, transportar y procesar flujos de datos a gran escala y con baja latencia.

El procesamiento de datos en tiempo real es un campo en constante evolución y las posibilidades son infinitas. Sigue explorando, experimentando y construyendo, y estarás bien encaminado para dominar las arquitecturas de datos modernas.

# Para detener y limpiar los contenedores de Docker
docker-compose down -v