Implementación de la Replicación Lógica en PostgreSQL: Multi-Maestro y Sincronización Selectiva de Datos

Introducción

La replicación lógica en PostgreSQL permite replicar tablas, filas o incluso columnas específicas entre bases de datos, habilitando patrones sofisticados de distribución de datos. A diferencia de la replicación física por streaming que copia todo el clúster de la base de datos, la replicación lógica proporciona un control granular sobre qué datos se replican y hacia dónde.

Replicación Lógica vs. Física

Replicación Física por Streaming

• Replica todo el clúster de la base de datos
• Replicación a nivel binario
• Réplicas de solo lectura
• Requiere la misma versión de PostgreSQL
• Menor sobrecarga (overhead)

Replicación Lógica

• Replicación selectiva de tablas/filas
• Compatible entre diferentes versiones
• Suscriptores con capacidad de escritura
• Mayor sobrecarga (overhead)
• Distribución de datos flexible

Casos de Uso para la Replicación Lógica

1. Distribución Selectiva de Datos: Replicar tablas específicas a diferentes regiones.
2. Configuraciones Multi-Maestro: Múltiples bases de datos con capacidad de escritura y sincronización bidireccional.
3. Actualizaciones entre Versiones: Replicar desde versiones antiguas a versiones nuevas de PostgreSQL.
4. Agregación de Datos: Consolidar datos de múltiples fuentes.
5. Cumplimiento de GDPR: Replicar solo columnas que no sean sensibles.

Prerrequisitos y Configuración

Requisitos de Configuración

En el Publicador (Origen):

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

En el Suscriptor (Destino):

# postgresql.conf
max_replication_slots = 10
max_logical_replication_workers = 10
max_worker_processes = 16

Reiniciar PostgreSQL

sudo systemctl restart postgresql

Configuración de Red

Asegúrese de que las bases de datos puedan comunicarse:

# Probar conexión desde el suscriptor al publicador
psql -h publisher.example.com -U replication_user -d source_db

Configurar pg_hba.conf en el publicador:

# Permitir conexiones de replicación
host    all    replication_user    subscriber_ip/32    md5

Configuración Básica de Replicación Lógica

Paso 1: Crear Usuario de Replicación

En el Publicador:

CREATE ROLE replication_user WITH REPLICATION LOGIN PASSWORD 'secure_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO replication_user;

Paso 2: Crear Tablas de Origen

En el Publicador:

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO users (username, email) VALUES 
    ('alice', '[email protected]'),
    ('bob', '[email protected]');

Paso 3: Crear Publicación

En el Publicador:

-- Publicar todas las tablas
CREATE PUBLICATION my_publication FOR ALL TABLES;

-- O publicar tablas específicas
CREATE PUBLICATION my_publication FOR TABLE users, orders;

-- O con filtros de fila (PostgreSQL 15+)
CREATE PUBLICATION active_users FOR TABLE users WHERE (status = 'active');

Ver publicaciones:

SELECT * FROM pg_publication;
SELECT * FROM pg_publication_tables;

Paso 4: Crear Tablas de Réplica

En el Suscriptor:

-- Las tablas deben tener una estructura idéntica
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

Paso 5: Crear Suscripción

En el Suscriptor:

CREATE SUBSCRIPTION my_subscription
    CONNECTION 'host=publisher.example.com port=5432 dbname=source_db user=replication_user password=secure_password'
    PUBLICATION my_publication;

Paso 6: Verificar Replicación

En el Publicador:

SELECT * FROM pg_stat_replication;
SELECT * FROM pg_replication_slots;

En el Suscriptor:

SELECT * FROM pg_stat_subscription;
SELECT * FROM users;  -- Debería ver los datos replicados

Configuración Avanzada

Replicación a Nivel de Columna (PostgreSQL 15+)

Replicar solo columnas específicas:

-- En el Publicador: Replicar solo columnas no sensibles
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    credit_card VARCHAR(20),  -- No será replicada
    created_at TIMESTAMP
);

CREATE PUBLICATION customer_basic 
    FOR TABLE customers (id, name, email, created_at);

Filtrado de Filas

Replicar solo registros activos:

CREATE PUBLICATION active_data 
    FOR TABLE orders WHERE (status IN ('pending', 'processing'));

Distribución regional de datos:

CREATE PUBLICATION us_customers 
    FOR TABLE customers WHERE (country = 'US');

CREATE PUBLICATION eu_customers 
    FOR TABLE customers WHERE (country IN ('UK', 'DE', 'FR'));

Múltiples Publicaciones

-- Publicador: Crear múltiples publicaciones
CREATE PUBLICATION oltp_data FOR TABLE users, orders;
CREATE PUBLICATION analytics_data FOR TABLE logs, metrics;

-- Suscriptor: Suscribirse a múltiples publicaciones
CREATE SUBSCRIPTION multi_sub
    CONNECTION 'host=publisher port=5432 dbname=mydb user=repuser password=pass'
    PUBLICATION oltp_data, analytics_data;

Replicación Bidireccional (Multi-Maestro)

Configurar Sincronización de Doble Vía

Configuración de Base de Datos A:

-- Crear publicación
CREATE PUBLICATION db_a_pub FOR TABLE shared_table;

-- Suscribirse a Base de Datos B
CREATE SUBSCRIPTION db_a_sub
    CONNECTION 'host=db-b.example.com dbname=mydb user=repuser'
    PUBLICATION db_b_pub
    WITH (origin = none);  -- Previene bucles de replicación

Configuración de Base de Datos B:

-- Crear publicación
CREATE PUBLICATION db_b_pub FOR TABLE shared_table;

-- Suscribirse a Base de Datos A
CREATE SUBSCRIPTION db_b_sub
    CONNECTION 'host=db-a.example.com dbname=mydb user=repuser'
    PUBLICATION db_a_pub
    WITH (origin = none);

Resolución de Conflictos

La replicación lógica utiliza "la última escritura gana" de forma predeterminada:

-- Establecer replica identity para rastrear conflictos
ALTER TABLE shared_table REPLICA IDENTITY FULL;

Estrategias para detección de conflictos:

1. Basada en marca de tiempo (Timestamp): Agregar columna updated_at

CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER update_shared_table_timestamp
    BEFORE UPDATE ON shared_table
    FOR EACH ROW
    EXECUTE FUNCTION update_timestamp();

2. Numeración de versiones:

CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    version INTEGER DEFAULT 1
);

Opciones de Sincronización Inicial de Datos

Opción 1: Copia Automática (Predeterminada)

-- El suscriptor copia automáticamente los datos existentes
CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);  -- Predeterminado

Opción 2: Sincronización Inicial Manual

Para conjuntos de datos grandes, use pg_dump:

# Volcar tablas específicas del publicador
pg_dump -h publisher.example.com -U postgres -d source_db \n    -t users -t orders --no-owner --no-acl > initial_data.sql

# Cargar en el suscriptor
psql -h subscriber.example.com -U postgres -d target_db < initial_data.sql

# Crear suscripción sin copia inicial
psql -h subscriber.example.com -U postgres -d target_db -c "
    CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=source_db user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = false);
"

Opción 3: Sincronización Inicial Paralela

-- Usar múltiples workers para una sincronización inicial más rápida
CREATE SUBSCRIPTION fast_sync
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (
        copy_data = true,
        streaming = on,
        synchronous_commit = off
    );

Monitoreo de la Replicación Lógica

Monitoreo del Publicador

-- Ver slots de replicación
SELECT 
    slot_name,
    plugin,
    active,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size
FROM pg_replication_slots;

-- Ver conexiones de replicación activas
SELECT 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    sync_state,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn)) AS send_lag
FROM pg_stat_replication;

Monitoreo del Suscriptor

-- Ver estado de la suscripción
SELECT 
    subname,
    pid,
    received_lsn,
    latest_end_lsn,
    pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
FROM pg_stat_subscription;

-- Comprobar errores de replicación
SELECT * FROM pg_stat_subscription WHERE last_msg_receipt_time < NOW() - INTERVAL '5 minutes';

Script de Monitoreo

#!/bin/bash
# logical-replication-monitor.sh

echo "=== Estado del Publicador ==="
psql -h publisher -d mydb -c "
    SELECT slot_name, active, 
           pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag
    FROM pg_replication_slots;"

echo ""
echo "=== Estado del Suscriptor ==="
psql -h subscriber -d mydb -c "
    SELECT subname, pid, 
           pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
    FROM pg_stat_subscription;"

Solución de Problemas

Problema 1: La suscripción no recibe datos

Verificar estado de suscripción:

SELECT subname, pid, subenabled, subconninfo FROM pg_subscription;

Habilitar suscripción si está deshabilitada:

ALTER SUBSCRIPTION my_sub ENABLE;

Buscar errores:

SELECT * FROM pg_stat_subscription;

Problema 2: El retraso (lag) de replicación está creciendo

Identificar tablas lentas:

SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del
FROM pg_stat_user_tables
ORDER BY (n_tup_ins + n_tup_upd + n_tup_del) DESC;

Aumentar procesos de workers:

# postgresql.conf en el suscriptor
max_logical_replication_workers = 20
max_worker_processes = 30

Problema 3: Hinchamiento (Bloat) del Slot de Replicación

Verificar uso del slot:

SELECT 
    slot_name,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

Eliminar slots inactivos:

SELECT pg_drop_replication_slot('inactive_slot_name');

Problema 4: Fallo en la Sincronización Inicial

Reiniciar sincronización inicial:

-- Eliminar y recrear suscripción
DROP SUBSCRIPTION my_sub;

CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);

Optimización de Rendimiento

1. Usar la Identidad de Réplica (Replica Identity) Adecuada

-- Predeterminado: Solo PRIMARY KEY
ALTER TABLE my_table REPLICA IDENTITY DEFAULT;

-- Full: Todas las columnas (mayor sobrecarga)
ALTER TABLE my_table REPLICA IDENTITY FULL;

-- Index: Usar un índice único específico
CREATE UNIQUE INDEX replica_idx ON my_table(col1, col2);
ALTER TABLE my_table REPLICA IDENTITY USING INDEX replica_idx;

2. Deshabilitar Restricciones Durante la Sincronización Inicial

-- Deshabilitar disparadores temporalmente para carga inicial más rápida
ALTER TABLE my_table DISABLE TRIGGER ALL;

-- Después de que se complete la sincronización inicial
ALTER TABLE my_table ENABLE TRIGGER ALL;

3. Aplicación Paralela (PostgreSQL 16+)

CREATE SUBSCRIPTION parallel_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (streaming = parallel);

Mejores Prácticas

1. Siempre use replica identity: Asegúrese de que las tablas tengan PRIMARY KEY o restricción UNIQUE.
2. Monitorear el retraso de replicación: Configure alertas para lag > 100MB o 5 minutos.
3. Usar filtros de fila con cuidado: Los filtros se evalúan en el publicador, afectando el rendimiento.
4. Planificar para conflictos: Implementar detección de conflictos en configuraciones multi-maestro.
5. Probar procedimientos de failover: Practique el cambio entre bases de datos.
6. Mantenimiento regular: Limpie los slots de replicación antiguos.

Conclusión

La replicación lógica proporciona capacidades poderosas de distribución de datos para PostgreSQL. Ventajas clave:

• Replicación selectiva de tablas y filas.
• Compatibilidad entre versiones.
• Configuraciones multi-maestro bidireccionales.
• Control granular de la privacidad de los datos.

Comience con una replicación unidireccional simple, pruebe a fondo y luego expanda a topologías más complejas según sea necesario.