Migración y Sincronización de Datos a Gran Escala con Apache Iceberg y Debezium

Introducción a la Sincronización Moderna en Big Data 🚀

En el ecosistema actual de datos, las organizaciones ya no dependen de un único monolito transaccional. La arquitectura moderna exige separar el almacenamiento analítico de los sistemas operacionales mediante la creación de Data Lakes robustos y escalables. Sin embargo, mover y mantener sincronizados terabytes o petabytes de datos desde múltiples bases de datos relacionales y NoSQL hacia un almacenamiento centralizado representa un desafío monumental.

Tradicionalmente, las cargas por lotes (batch ETL) nocturnas provocaban latencias inaceptables y picos de consumo en los sistemas de producción. La solución moderna es la combinación de Change Data Capture (CDC) mediante Debezium y la gestión avanzada de formatos de tabla con Apache Iceberg. Esta combinación permite mantener un lago de datos actualizado en tiempo real sin impactar el rendimiento operacional.

Arquitectura de la Solución 📊

Para entender cómo fluyen los datos desde una base de datos relacional hasta un almacenamiento analítico optimizado, utilizaremos una arquitectura basada en eventos distribuidos.

Componentes Clave del Pipeline

Paso 1: Configuración del Origen y Captura con Debezium 🛠️

El primer paso consiste en configurar nuestra base de datos de origen (por ejemplo, PostgreSQL) para que emita sus registros de transacciones y permitir que Debezium los capture sin modificar el esquema de las tablas de negocio.

Configuración de PostgreSQL para CDC

Debemos asegurarnos de que PostgreSQL esté configurado con el nivel de replicación adecuado para emitir eventos lógicos (logical decoding).

# postgresql.conf
wal_level = logical
max_wal_senders = 4
max_replication_slots = 4

Una vez configurada la base de datos, desplegamos el conector de Debezium en Kafka Connect utilizando una petición REST para iniciar la captura de tablas específicas.

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "inventory",
    "database.server.name": "dbserver1",
    "table.include.list": "public.customers,public.orders",
    "plugin.name": "pgoutput"
  }
}

Paso 2: Consumo y Escritura en Apache Iceberg ⚡

Una vez que los eventos de inserción, actualización y eliminación (INSERT, UPDATE, DELETE) fluyen por los tópicos de Kafka, necesitamos un motor de procesamiento que escriba estos datos en nuestro Data Lake utilizando el formato Apache Iceberg. Iceberg nos aporta garantías ACID, control de versiones y evolución de esquemas sin corromper los archivos históricos.

Procesamiento con PySpark y Apache Iceberg

Utilizaremos PySpark para leer los streams de Kafka y aplicar las operaciones de upsert directamente sobre las tablas Iceberg.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType, IntegerType, TimestampType

# Inicializar Spark con soporte para Iceberg
spark = SparkSession.builder \
    .appName("DebeziumToIceberg") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog") \
    .config("spark.sql.catalog.spark_catalog.type", "hive") \
    .getOrCreate()

# Definir esquema del mensaje Debezium
schema = StructType() \
    .add("before", StringType()) \
    .add("after", StringType()) \
    .add("op", StringType()) \
    .add("ts_ms", IntegerType())

# Leer stream desde Kafka
df_kafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "dbserver1.public.customers") \
    .load()

# Procesar y escribir en tabla Iceberg
def write_to_iceberg(batch_df, batch_id):
    # Filtrar y parsear datos del payload
    parsed_df = batch_df.select(from_json(col("value").cast("string"), schema).alias("data")) \
                        .select("data.*");
                        
    # Aplicar lógica de upsert usando Iceberg MERGE INTO
    parsed_df.createOrReplaceTempView("changes")
    
    spark.sql("""
        MERGE INTO spark_catalog.default.customers t
        using changes s
        ON t.id = s.after.id
        WHEN MATCHED AND s.op = 'u' THEN UPDATE SET *
        WHEN MATCHED AND s.op = 'd' THEN DELETE
        WHEN NOT MATCHED AND s.op = 'c' THEN INSERT *
    """)

# Ejecutar el stream continuo
query = df_kafka.writeStream \
    .foreachBatch(write_to_iceberg \
    .outputMode("update") \
    .start()

query.awaitTermination()

Paso 3: Optimización y Mantenimiento del Data Lake 🧹

Un pipeline de sincronización continua genera una gran cantidad de archivos pequeños (small files problem) debido a las actualizaciones frecuentes. Para mantener la salud y el rendimiento de las consultas sobre Apache Iceberg, es necesario programar tareas de mantenimiento regulares.

Tareas de Mantenimiento Esenciales

Ejemplo de Mantenimiento con Spark SQL

-- Compactar archivos pequeños en la tabla Iceberg
CALL spark_catalog.system.rewrite_data_files(table => 'default.customers');

-- Expirar snapshots antiguos con más de 7 días de antigüedad
CALL spark_catalog.system.expire_snapshots(table => 'default.customers', older_than => TIMESTAMP '2023-10-01 00:00:00.000');

-- Eliminar archivos huérfanos que no están enlazados a ningún snapshot
CALL spark_catalog.system.remove_orphan_files(table => 'default.customers');

Preguntas Frecuentes (FAQ) 🙋‍♂️

Conclusión y Próximos Pasos 🎯

La integración de Debezium y Apache Iceberg representa el estado del arte en la sincronización y gestión de arquitecturas de Big Data modernas. Al eliminar los procesos por lotes tradicionales y adoptar flujos basados en CDC con almacenamiento transaccional abierto, las organizaciones logran una consistencia analítica cercana al tiempo real.

Te invitamos a probar esta arquitectura en tu entorno local usando contenedores de Docker Compose para simular el clúster de Kafka, PostgreSQL y Spark de forma integrada.