Cómo Sincronizar SQL Server con Apache Kafka Connect en Tiempo Real: Replicación CDC sin Bloquear tu Base de Datos

El Antipatrón de Producción: Por Qué Replicar con Apache Spark y Queries SELECT Congeló Nuestra Aplicación

En proyectos de ingeniería de datos a escala empresarial, existe un error de diseño tan recurrente como devastador: tratar a una base de datos transaccional (OLTP) de misión crítica como si fuera un data lake analítico.

Durante una de nuestras intervenciones en un cliente corporativo, el equipo de analítica necesitaba replicar continuamente los cambios de tablas maestras de Microsoft SQL Server hacia su plataforma de datos. La solución inicial parecía sencilla sobre el papel: levantaron un servicio programado sobre Apache Spark que ejecutaba micro-batches cada 2 minutos lanzando consultas SELECT * FROM tabla WHERE updated_at > ? mediante JDBC.

1. La trampa del polling periódico en bases de datos transaccionales

En entornos de desarrollo con diez mil filas, este enfoque de *polling* funciona sin aparentes contratiempos. Sin embargo, en el entorno de producción con millones de registros y cientos de transacciones por segundo, las consecuencias fueron inmediatas:

• Escalado y contención de bloqueos (Lock Escalation): Para resolver las consultas analíticas masivas de Spark, SQL Server bloqueaba páginas de datos completas que rápidamente escalaban a bloqueos exclusivos de tabla (Table Locks).
• Agotamiento del Connection Pool: Cada executor de Spark abría conexiones JDBC concurrentes que competían directamente con el pool de conexiones de la API de backend que utilizaban los clientes finales.
• Table Scans devastadores: Si las columnas updated_at carecían de índices perfectamente alineados o contenían registros con transacciones en vuelo, el motor forzaba barridos completos de tabla, elevando la CPU del servidor al 100%.

2. Síntomas críticos: cuando la réplica tira abajo el core de negocio

El resultado fue crítico: la aplicación web y móvil de los clientes comenzó a congelarse. Las peticiones HTTP transaccionales (pagos, altas de usuarios, checkout) entraban en timeout esperando la liberación de locks retenidos por las consultas de Spark. Intentar arreglarlo bajando los niveles de aislamiento a READ UNCOMMITTED (o NOLOCK) solo introdujo lecturas sucias (dirty reads) y datos duplicados en el lago analítico.

Para resolver este problema sin poner en riesgo la disponibilidad del negocio, fue necesario aplicar un cambio radical de arquitectura: desacoplar la ingesta de la capa de ejecución SQL mediante Change Data Capture (CDC) implementado sobre Kafka Connect.

La Solución Arquitectónica: Change Data Capture (CDC) y Kafka Connect sin Impacto Operativo

La diferencia fundamental entre el *polling* tradicional y el *Change Data Capture* basado en logs radica en dónde y cómo se leen las modificaciones.

1. Lectura del Transaction Log vs Consultas Activas a Tablas

En cualquier base de datos relacional compatible con ACID (como SQL Server o PostgreSQL gestionado en arquitectura de alta disponibilidad y resiliencia en Cloud SQL), toda operación INSERT, UPDATE o DELETE se escribe primero en un registro secuencial persistente en disco: el Transaction Log (archivo .ldf) o Write-Ahead Log (WAL), antes de confirmarse en las páginas de datos en memoria.

Un conector de CDC moderno no ejecuta sentencias SQL contra las tablas vivas. En su lugar, lee de forma pasiva y asíncrona los bytes que el propio motor de la base de datos ya ha serializado en su log de transacciones. Esto garantiza tres ventajas arquitectónicas cruciales:

1. Cero Contención de Locks: No compite por bloqueos de lectura ni bloquea escrituras de los usuarios finales.
2. Captura Fidedigna de DELETE: Un polling periódico jamás detecta qué registro fue borrado entre dos ejecuciones a menos que exista un soft-delete; el CDC captura cada evento de supresión de forma nativa.
3. Baja Latencia de Replicación (Sub-segundo): En cuanto se confirma el commit, el evento viaja a través de Kafka Connect hacia los tópicos de Apache Kafka en milisegundos.

2. Anatomía de Debezium Connector para SQL Server

Debezium es la plataforma open-source de referencia para CDC construida sobre el runtime de Kafka Connect. En el caso de SQL Server, Debezium aprovecha el feature nativo de Change Data Capture del motor de Microsoft, leyendo los datos desde las tablas de captura internas gestionadas por el proceso de captura de SQL Server Agent, traduciéndolos a estructuras enriquecidas que incluyen el estado anterior (before), el estado posterior (after) y los metadatos de la transacción (timestamp, LSN y tipo de operación).

Matriz Comparativa: Polling JDBC vs Apache Spark Batch vs Log-Based CDC con Kafka Connect

Antes de seleccionar la tecnología de ingesta, los arquitectos de datos deben contrastar las compensaciones (*trade-offs*) operativas de cada enfoque:

Vídeo Tutorial: Demostración Práctica de Replicación de SQL Server a Apache Kafka

En el siguiente tutorial práctico del canal oficial de Google Cloud con Eduardo, puedes ver en detalle el despliegue del entorno, la verificación de los conectores y la captura en tiempo real de transacciones directamente desde la terminal:

Despliegue Paso a Paso: Entorno Docker con Apache Kafka, KRaft y Kafka Connect

A continuación, implementaremos un entorno completo y reproducible en Docker que integra Microsoft SQL Server 2022 con CDC habilitado, un broker de Apache Kafka ejecutándose con el moderno consenso KRaft (sin ZooKeeper) y un nodo distribuido de Kafka Connect con el plugin oficial de Debezium SQL Server. Si deseas profundizar en la configuración de la red de brokers, consulta nuestra guía sobre cómo desplegar Apache Kafka en Docker con KRaft.

1. Preparación de SQL Server: Habilitar CDC en Base de Datos y Tablas

Para que el motor de SQL Server comience a registrar los cambios en sus tablas de captura internas, debemos ejecutar los siguientes procedimientos almacenados en la base de datos objetivo (por ejemplo, EmpresaDB). Asegúrate de que el servicio SQL Server Agent se encuentre en estado de ejecución:

-- 1. Habilitar CDC a nivel de base de datos
USE EmpresaDB;
GO

EXEC sys.sp_cdc_enable_db;
GO

-- 2. Verificar que CDC quedó habilitado correctamente
SELECT name, is_cdc_enabled 
FROM sys.databases 
WHERE name = 'EmpresaDB';
GO

-- 3. Crear tabla de ejemplo para transacciones de clientes
CREATE TABLE Clientes (
    cliente_id INT IDENTITY(1,1) PRIMARY KEY,
    nombre NVARCHAR(100) NOT NULL,
    email NVARCHAR(150) NOT NULL,
    pais VARCHAR(50) DEFAULT 'España',
    fecha_registro DATETIME DEFAULT GETDATE()
);
GO

-- 4. Habilitar CDC para la tabla específica
EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name   = N'Clientes',
    @role_name     = NULL, -- Permite acceso a todos los autorizados
    @supports_net_changes = 1;
GO

-- 5. Comprobar estado de la tabla bajo seguimiento CDC
SELECT name, is_tracked_by_cdc 
FROM sys.tables 
WHERE name = 'Clientes';
GO

2. Docker Compose: Kafka Broker (KRaft) + Kafka Connect con Debezium

Definimos el archivo docker-compose.yml. Observa que utilizamos la imagen oficial de debezium/connect, la cual ya incluye preinstalados los conectores para SQL Server, PostgreSQL, MySQL y Oracle:

services:
  # Broker de Apache Kafka con KRaft (Sin ZooKeeper)
  kafka:
    image: apache/kafka:3.9.0
    container_name: kafka-broker
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,PLAINTEXT_HOST://:29092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    networks:
      - cdc-network

  # Motor de Kafka Connect con Plugins de Debezium
  kafka-connect:
    image: quay.io/debezium/connect:2.7
    container_name: kafka-connect
    depends_on:
      - kafka
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
    networks:
      - cdc-network

networks:
  cdc-network:
    driver: bridge

Inicia el entorno en segundo plano ejecutando:

docker compose up -d

3. Configuración del Conector Debezium mediante REST API y SMT

Una vez levantado el contenedor de Kafka Connect (puedes verificar su disponibilidad en el puerto 8083), registramos el conector enviando un payload JSON mediante la API REST.

Punto Crítico de Producción: De forma predeterminada, Debezium emite una estructura JSON compleja con sobres que contienen schema, before y after. Para que los sistemas downstream reciban una estructura limpia y plana, aplicamos la transformación SMT (Single Message Transform) denominada io.debezium.transforms.ExtractNewRecordState:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d '{
    "name": "sqlserver-cdc-clientes",
    "config": {
      "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
      "tasks.max": "1",
      "database.hostname": "sqlserver",
      "database.port": "1433",
      "database.user": "sa",
      "database.password": "TuPasswordSeguro123!",
      "database.names": "EmpresaDB",
      "topic.prefix": "cdc_sqlserver",
      "table.include.list": "dbo.Clientes",
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
      "schema.history.internal.kafka.topic": "schema-changes.clientes",
      
      "transforms": "unwrap",
      "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
      "transforms.unwrap.drop.tombstones": "false",
      "transforms.unwrap.delete.handling.mode": "rewrite"
    }
  }'

4. Verificación del Streaming de Eventos en Terminal

Comprobamos el estado del conector para confirmar que la tarea se encuentra en estado RUNNING:

curl -s http://localhost:8083/connectors/sqlserver-cdc-clientes/status | jq .

Para auditar los eventos en tiempo real a medida que se insertan registros en SQL Server, abrimos un consumidor por consola utilizando los comandos de Apache Kafka CLI:

docker exec -it kafka-broker /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic cdc_sqlserver.EmpresaDB.dbo.Clientes \
  --from-beginning

Si ejecutas un INSERT INTO Clientes (nombre, email) VALUES ('Eduardo Martinez', 'eduardo@ejemplo.com'); en SQL Server, verás aparecer instantáneamente en tu terminal el evento deserializado en formato JSON con latencia inferior a 50 milisegundos.

Alertas de Producción, FinOps y Antipatrones en Replicación CDC

Para mitigar este riesgo operativo en producción:

• Monitorizar el Log Reuse Wait: Ejecuta periódicamente SELECT name, log_reuse_wait_desc FROM sys.databases;. Si el valor devuelve REPLICATION durante períodos prolongados, significa que el conector está atrasado o desconectado.
• Alertas de Consumer Lag en Kafka Connect: Monitoriza la métrica JMX source-record-poll-rate y el lag respecto al último LSN (Log Sequence Number) para disparar avisos antes de que el disco alcance el 80% de su capacidad.

Preguntas Frecuentes sobre la Sincronización de SQL Server con Kafka Connect (FAQ)