- Para transmitir cambios de Postgres a otros sistemas en tiempo real, se necesita CDC (Change Data Capture), y entre una notificación simple y la replicación basada en WAL hay grandes diferencias tanto en confiabilidad como en carga operativa
- Listen/Notify es la forma más ligera de empezar, pero por su semántica de entrega at-most-once, sus notificaciones temporales y su límite de payload de 8000 bytes, se parece más a una señal auxiliar que a un CDC central
- El polling de tablas y las tablas de auditoría (outbox pattern) pueden implementarse solo con tablas y triggers estándar, pero hay que resolver por cuenta propia la detección de borrados, los diff, el orden de commit, la amplificación de escritura y el backpressure
- La replicación lógica (logical replication) es una forma potente de hacer streaming de insert/update/delete desde el WAL, pero la aplicación debe encargarse del replication slot, los ack, los reinicios y la respuesta al volumen de procesamiento
- Sequin, basado en la replicación lógica de Postgres, transmite cambios hacia SQS, Kafka, Elasticsearch, Redis, endpoints HTTP y otros destinos, reduciendo la carga de tener que manejar directamente un replication slot
Cuándo necesitas CDC en Postgres
- Postgres es muy fuerte para manejar datos almacenados, pero si quieres disparar workflows a partir de cambios en tablas o hacer streaming en tiempo real hacia otros almacenes de datos, sistemas o servicios, tienes que diseñar por separado el movimiento de datos
- Change Data Capture (CDC) es una forma de identificar y capturar cambios en la base de datos para luego entregarlos en tiempo real a sistemas downstream
- Hay varias formas de detectar cambios en Postgres, y difieren en dificultad de implementación, confiabilidad y carga operativa
Listen/Notify: el pub-sub más simple
- Listen/Notify de Postgres es una función de comunicación entre procesos y funciona con el patrón publish-subscribe
- Una sesión puede hacer
listensobre un canal específico, y la actividad de la base de datos u otras sesiones pueden enviarnotifya ese canal - Para capturar cambios puede usarse junto con triggers
- Un trigger de ejemplo, en
after insert or update or delete, construye un JSON contable,idyactiondel registro modificado y llama apg_notify('table_changes', payload::text)
- Un trigger de ejemplo, en
- Sus límites son claros
- Tiene semántica de entrega at-most-once, y el listener debe estar conectado en el momento en que se emite la notificación
- El listener solo recibe notificaciones a partir del momento en que se suscribe, así que incluso una desconexión breve por un problema de red puede hacer que se pierdan avisos
- El tamaño del payload está limitado a 8000 bytes, y si se supera, el comando
notifyfalla - El tamaño del payload también incluye el nombre del canal, y como los identificadores de Postgres, ese nombre puede tener hasta 64 bytes
- Puede servir para detección básica de cambios o para optimizar el polling de tablas, pero no encaja tan bien en requisitos de CDC más complejos
Polling de tablas: simple, pero débil para borrados y diff
- La forma más simple y robusta de capturar cambios es hacer polling directamente sobre la tabla
- Cada tabla necesita una columna como
updated_atque se actualice cada vez que cambia una fila, y si hace falta puede crearse con un trigger - La combinación de
updated_atconidpuede usarse como cursor, y la lógica de la aplicación se encarga de guardar y administrar ese cursor - Si además se usa una suscripción a Notify, la aplicación puede enterarse de que se insertó o modificó un registro y reducir la frecuencia de polling
- Como las notificaciones de Postgres son temporales, conviene usarlas solo como optimización sobre el polling
- Tiene tres desventajas principales
- Como las filas borradas ya no permanecen en la tabla, no es posible detectar borrados
- Como alternativa, un trigger de delete puede guardar el
idy las columnas necesarias en una tabla aparte comodeleted_contacts, que la aplicación puede consultar por polling - Se puede saber que un registro fue actualizado, pero no qué cambió
- Como en Postgres los datetime y las secuencias pueden desalinearse del orden de commit, al leer bloques según
updated_atse pueden omitir filas que todavía estaban en proceso de commit
- Es una opción razonable para seguimiento simple de cambios, donde los borrados, los diff y las omisiones ocasionales no sean un problema grande
Tabla de auditoría: guardar el log de cambios con el outbox pattern
- El enfoque de tabla de auditoría (audit table) registra los cambios en una tabla separada
changelog, y también se conoce como outbox pattern changelogpuede tener columnas relacionadas con el cambioaction: si fueinsert,updateodeleteold: eljsonbdel registro antes del cambio; en insert queda vacíovalues: eljsonbde los campos modificados; en delete queda vacíoinserted_at: el momento en que ocurrió el cambio
- Para implementarlo hace falta una función trigger que inserte en
changelogcada vez que ocurra un cambio, además de un trigger por cada tabla que se quiera vigilar - También es posible consumir
changelogcomo si fuera una cola- Un worker de la aplicación toma cambios desde la tabla
- Para un procesamiento aproximadamente exactly-once, se puede usar
for update skip lockedde Postgres - El worker puede abrir una transacción, bloquear un lote con
order by timestamp limit 100 for update skip locked, procesarlo, borrar los registros procesados y luego hacer commit
- Tiene desventajas operativas
- Una sola escritura en la tabla original genera varias escrituras en la tabla de auditoría, lo que produce amplificación de escritura (write amplification)
- Normalmente se generan al menos tres escrituras: el insert inicial en la tabla de auditoría, un update durante el procesamiento y un delete después de procesarlo
- Si quieres hacer fan-out con workers, tienes que diseñarlo tú mismo según tu aplicación
- Antes de desplegar a escala de producción, probablemente haya que ajustar la función trigger y el diseño de tablas
- También puede ser necesario considerar políticas detalladas, como el tiempo máximo que un worker puede retener un cambio tomado de la cola
- Si el worker no logra procesar con éxito, la tabla de auditoría sigue creciendo, así que la gestión del backpressure es limitada
Foreign Data Wrapper: una opción más cercana a sincronizar Postgres específicos
- Foreign Data Wrapper (FDW) es una función que permite leer y escribir fuentes de datos externas desde una base de datos Postgres
- La extensión basada en FDW con soporte más extendido es
postgres_fdw- Permite conectar dos bases de datos Postgres y crear una estructura parecida a una view donde una base puede referenciar tablas de la otra
- Internamente, una base de datos Postgres actúa como cliente y la otra como servidor
- Cuando consultas una foreign table, la base cliente envía consultas a la base servidor mediante el wire protocol de Postgres
- FDW no es una forma común de capturar cambios y fuera de situaciones muy específicas es difícil recomendarlo
- Si lo que quieres es escribir cambios de una base de datos Postgres en otra base de datos Postgres, FDW puede encajar
- Un ejemplo sería usar una base de datos separada para contabilidad y otra para la aplicación
- En vez de una etapa intermedia de captura de cambios, puedes reflejar los cambios directamente entre bases con
postgres_fdw
- También es posible crear un FDW propio que haga POST de los cambios hacia una API interna
- Como escribe en la API dentro del commit, la API puede rechazar el cambio y hacer rollback del commit
- FDW es potente, pero rara vez es la mejor opción para CDC, y escribir un FDW propio se acerca a ser el trabajo más grande entre estas estrategias de captura de cambios
- Crear un FDW propio se ha vuelto más fácil con herramientas como Supabase wrappers, pero sigue siendo un trabajo importante
Replicación lógica directa: CDC potente basado en WAL
- Postgres tiene un protocolo para replicación de base de datos, y uno de sus modos es la replicación lógica (logical replication)
- La replicación lógica está construida sobre el WAL (write-ahead log) de Postgres
- Se rastrean todos los insert, update y delete de la base de datos
- Los cambios se envían por streaming al subscriber
- El usuario primero crea un replication slot en el primary
- Se usa una forma como
pg_create_logical_replication_slot('<your_slot_name>', '<output_plugin>')
- Se usa una forma como
output_pluginindica qué plugin se usará para decodificar los cambios del WALpgoutputes el plugin por defecto y produce salida en el formato binario que espera el servidor clientetest_decodinges un plugin de salida simple que entrega los cambios del WAL en una forma legible para humanos- Aunque no es un plugin integrado en Postgres,
wal2jsones un plugin popular, y JSON suele ser más fácil de manejar como punto de partida para una aplicación que el formato binario de Postgres
- Después de crear el replication slot, ya se puede iniciar y consumir
- El replication slot usa una parte del protocolo de Postgres distinta de la de las consultas estándar
- Varias librerías cliente ofrecen funciones para trabajar con replication slots
- Un ejemplo con
psycopg2consume mensajes WAL concursor.start_replication(...)ycursor.consume_stream(...), y envía ack concursor.send_feedback(flush_lsn=msg.wal_end)
- El cliente debe hacer ack de los mensajes WAL recibidos, y el replication slot funciona de forma parecida a Kafka con offsets
- La replicación lógica es una forma robusta creada para CDC, pero es compleja
- Los replication slots y el protocolo de replicación son menos familiares para los desarrolladores que las tablas y consultas normales
- Hace falta una estrategia para no perder mensajes durante reinicios
- Hay que diseñar el sistema para manejar el gran volumen de mensajes que puede salir de Postgres
Sequin: una herramienta CDC que envuelve la replicación lógica
- Sequin es una herramienta de CDC que entrega cambios y filas de Postgres hacia colas, streams, índices de búsqueda, cachés, endpoints HTTP y otros destinos
- Entre los destinos están SQS, Kafka, Elasticsearch, Redis y endpoints HTTP
- Sequin usa internamente la replicación lógica de Postgres, pero abstrae la complejidad del protocolo de bajo nivel
- Puede capturar insert, update y delete, y en update y delete captura tanto el valor
newcomo el valoroldde la fila - Estas son condiciones en las que conviene considerar Sequin
- Necesitas CDC en tiempo real
- Quieres hacer streaming directo a un destino como SQS o un webhook sin pasar por un sistema intermedio
- Necesitas funciones como backfill de datos históricos y filtrado de cambios con cláusulas SQL
where - Necesitas una alternativa más simple que manejar directamente un replication slot
- Necesitas garantías de procesamiento exactly-once
- También tiene desventajas
- Sequin no es una extensión interna de Postgres, sino una herramienta de terceros que corre junto a la base de datos
- Justamente porque no es una extensión, tiene amplia compatibilidad con cualquier base de datos Postgres, pero si no usas Sequin Cloud, tendrás que montar infraestructura adicional por tu cuenta
Criterios de elección
- En la etapa inicial, Listen/Notify y el polling de tablas son adecuados
- Listen/Notify sirve bien para capturar eventos no críticos, hacer prototipos y optimizar el polling
- El polling es una solución directa y razonable para casos de uso simples
- En una etapa un poco más seria, la tabla de auditoría puede ser una opción intermedia
- Puede capturar payloads
newyoldde la fila - Si se implementa bien, se puede obtener un sistema de procesamiento exactly-once
- Al escalar, la amplificación de escritura y la falta de backpressure se vuelven un problema, y en una configuración manual un error puede hacer que se pierdan mensajes
- Puede capturar payloads
- En la etapa de escalamiento, la replicación lógica es lo más cercano a una solución robusta
- Aun así, se recomienda usar una herramienta como Sequin en lugar de leer directamente desde el slot
- FDW es una función interesante, pero es poco probable que resuelva necesidades generales de CDC
1 comentarios
Opiniones de Hacker News
Triggers + tablas de historial (tablas de auditoría) son la respuesta correcta en el 98% de los casos. Si todavía no las estás usando, puedes empezar hoy. Es una técnica comprobada desde hace más de 30 años.
Hay un ejemplo simple de implementación genérica en https://gist.github.com/slotrans/353952c4f383596e6fe8777db5d.... Es un enfoque que sacrifica eficiencia de espacio a cambio de una “implementación fácil”.
Si puedes almacenar datos inmutables, sería ideal, pero probablemente en la base de datos tengas una enorme cantidad de datos mutables y todos los días estés olvidando muchas cosas. No lo olvides: usa tablas de historial.
Referencia: https://github.com/matthiasn/talk-transcripts/blob/master/Hi...
Es mejor no usar bibliotecas o técnicas de seguimiento de historial en la capa de aplicación, como Papertrail. Son lentas y propensas a errores, y no capturan cambios en la DB que pasen por fuera del stack de la app. Intentar poner timestamps
updateddesde la app también es fundamentalmente incorrecto, porque cada servidor web tiene un reloj distinto. Hay que usar el reloj de la DB, y ese es el único reloj correcto.now()dentro de la consulta.Pero sincronizar solo con este timestamp no es suficiente. Esto se debe a que el timestamp se genera en el inicio de la transacción, no en el momento del commit.
Si haces polling de una tabla filtrando por timestamps recientes, podrías perder parte de transacciones cuyo orden de commit quedó intercalado. Puedes poner una ventana de margen consultando unos minutos más hacia atrás y eliminar duplicados, pero en PostgreSQL las transacciones pueden durar un tiempo ilimitado, y consultar demasiado hacia atrás genera mucho desperdicio. Si la exactitud y la eficiencia importan, este enfoque no es el adecuado.
Incluye número de secuencia del log, hora de la DB y, si se usa
REPLICA IDENTITY FULL, incluso el estado antes/después de los cambios. Luego, al materializar colecciones en lugares como Snowflake, obtienes por defecto tablas sincronizadas que siguen las actualizaciones de la DB fuente.Desde el mismo data lake base también se puede transformar o materializar el historial completo de tablas con fines de auditoría, sin necesidad de volver a conectar otro capturador o lector de WAL a la DB fuente.
También dejé explicada por separado una forma en SQLite que implementa un patrón similar basado en columnas, no en JSON: https://simonwillison.net/2023/Apr/15/sqlite-history/
Este artículo resume bien y de forma concisa varios enfoques posibles usando funcionalidades básicas de Postgres.
En la parte de “capturar cambios en una tabla de auditoría”, en una empresa anterior usamos con buenos resultados el patrón de Temporal Tables. A diferencia de otros RDBMS importantes, Postgres no lo trae integrado, pero existe un patrón simple que se puede usar mediante funciones SQL: https://github.com/nearform/temporal_tables
Permite ver el estado de una tabla en un momento determinado, para responder preguntas como “¿cuáles eran las configuraciones de este usuario el 12 de agosto?”, “¿cuántos registros pendientes había anoche a las 11:55?” o “muéstrame la diferencia entre los feature flags actuales y los de hace una semana”.
Hace un tiempo hice consultoría para una empresa que tenía un SQL Server monolítico enorme. No era Postgres, pero habría sido parecido si lo hubiera sido.
Llevaba décadas en operación y se usaba para todo tipo de fines dentro de la empresa; en la práctica, todas las aplicaciones y procesos de negocio de toda la compañía guardaban datos en esa base de datos.
El problema era que había muchas aplicaciones consultando esa DB, y también una cantidad enorme de procesos y procedimientos que insertaban y modificaban datos, así que cuando cambiaban o se agregaban procesos aguas arriba de inserción/modificación, terminaban rompiendo invariantes a nivel de aplicación. Incluso los procesos normales se comportaban distinto si había datos malos.
Era muy difícil rastrear la causa, porque las cosas que revisábamos casi siempre habían sido escritas 10 años antes y esas personas ya se habían ido de la empresa.
Me pregunto si se podrían capturar los cambios en una base de datos Postgres en alguna forma de DAG, para saber qué procesos insertan, modifican o eliminan datos y cómo se han comportado históricamente, cómo varias aplicaciones consultan esos datos y cómo las estadísticas de consultas cambian con el tiempo.
No sé si hay precedentes para esto ni qué enfoque permitiría construir una herramienta así. Hace tiempo pensé en crear algo parecido, pero parece un área en la que hace falta un nivel de entendimiento similar al de un ingeniero del core de Postgres para tomar buenas decisiones.
No se obtiene información de origen a nivel de cliente para cada cambio.
Aun así, hay formas de rodearlo. El stream de replicación lógica también puede incluir mensajes informativos de la función
pg_logical_emit_message, así que el cliente puede insertar metadatos directamente. Tal vez se podría configurar para emitir un identificador de cliente al inicio de cada transacción.last updated by). Podría ser un antipatrón, así que me gustaría una solución más robusta.Este enfoque funciona en casi cualquier solución de la familia SQL que use WAL o triggers.
En SQL Server he usado varias veces el enfoque con triggers, pero registrar todas las consultas tiende a volverlo lento. Diseñar un mecanismo de inserción que no bloquee la operación no es perfecto, y puede hacer falta muestreo.
Si vas por la ruta de las “tablas de auditoría”, basta con usar pgaudit. Es una extensión probada en producción y, si usas AWS, también está disponible en RDS.
https://github.com/pgaudit/pgaudit/blob/master/README.md
https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appen...
https://github.com/arkhipov/temporal_tables
https://news.ycombinator.com/item?id=26748096
No hace falta hacerlo. Querer esto significa convertir las relaciones de Postgres en contratos. Ningún servicio podría persistir estado interno.
Tal vez sea posible si realmente te comprometes con el diseño guiado por dominio, pero es mejor usar un sistema basado en eventos que sea liviano y práctico.
Cualquier cosa basada en eventos es 1000 veces más compleja.
El enfoque de hacer polling sobre una columna
updated_atno es robusto en su forma más simple, porque no hay garantía de que las transacciones hagan commit en ese orden.updated_atde la fila 1 queda en2023-09-22 12:00:01.Un momento después empieza la transacción B, el
updated_atde la fila 2 queda en2023-09-22 12:00:02, y B hace commit primero.Se ejecuta la consulta de polling, ve la fila 2 como el cambio más reciente y actualiza el cursor a
2023-09-22 12:00:02; luego, si A hace commit después, se pierde la fila 1.Una forma simple de evitar este problema es no hacer polling casi en tiempo real. El orden termina alineándose de forma consistente.
Una propuesta más robusta podría ser usar una secuencia. Por ejemplo, tener una columna
updated_at_idxque se incremente cada vez que cambia una fila.Me pregunto si, usando un trigger before que inserte
now(), los timestampsupdated_atde las dos filas podrían diferir del orden de commit de las transacciones.updated_aty el timestamp de commit no tienen por qué ser iguales, peroupdated_atdebería representar con precisión el orden de commit a nivel de milisegundos o microsegundos._txid, que el trigger establece con el ID de transacción actual, en vez deupdated_at. Luego, al hacer polling, usamostxid_current()para comprobar qué transacciones hicieron commit y cuáles todavía no.Es un poco delicado y es muy fácil cometer errores en los valores límite, pero lleva años funcionando bien en producción.
Excelente artículo.
Si usas Elixir y Postgres, hice una pequeña librería que escucha cambios del WAL con un enfoque parecido: https://github.com/cpursley/walex
Todos estos enfoques son medio flojos y, personalmente, creo que el polling es lo más práctico
Me gustaría que Postgres innovara en esta área
Hasta que entren en el estándar SQL, creo que será difícil que ganen impulso dentro del espacio del kernel de los DBMS relacionales. Las opciones son muchas y complejas, y las soluciones exitosas en espacio de usuario tampoco suelen implicar una carga excesiva en términos de rendimiento
Como referencia, quienes investigan este campo en general se inclinan por el enfoque de tablas de auditoría. Porque mantiene propiedades ACID consistentes dentro de la base de datos y conserva a Postgres como único punto de falla, en lugar de agregar proxies o trabajos de polling
Hay un gran vacío en el mundo de los datos. En vez de preguntarle al almacén de datos por los resultados, estaría bueno que los resultados de las consultas se enviaran incrementalmente por push
Hago mucho análisis en tiempo real y de streaming; se puede procesar streams y también resolver parte de eso con vistas materializadas dentro del almacén de datos. Pero una vez que los datos entran en la DB o en el data lake, para ver cambios aguas abajo en la práctica se vuelve al polling
Si quieres reaccionar cuando ocurre cierta situación en los datos, o actualizar una pantalla sin recargar la página, no hay muchas soluciones limpias. Las soluciones de este artículo también parecen más rodeos que funciones de primera clase
Si quieres crear un reporte que se actualice en tiempo real sin recargar la página, normalmente terminas cargando datos desde la DB y luego enviando los cambios a la GUI con Kafka y WebSocket. Así acabas operando una extraña arquitectura lambda, donde parte del análisis se hace en código y parte en la DB
Hay innovación en esta área. KSQL y Kafka Streams pueden emitir cambios, Materialize tiene suscripciones y ClickHouse tiene vistas en vivo. Pero muchas funciones son nuevas o están en etapa de preview, y no encajan del todo. Las probé todas, pero sentí que le trasladan demasiado trabajo al desarrollador
Sería bueno tener una biblioteca que permita recibir directamente un feed de cambios con una opción como
[select * from orders with suscribe]. Es un área lo suficientemente importante, pero ha recibido menos atención de la que mereceHay una gran trampa de la replicación que el artículo no cubre, y por eso no uso replicación
Postgres intenta garantizar con mucha fuerza que los consumidores de slots de replicación no se pierdan datos. Por eso, si un consumidor no consume datos del slot, Postgres sigue conservando amablemente los datos no leídos, hasta que finalmente el disco se llena y la DB se cae. Me pasó durante prototipado en dos DB SaaS distintas, y la única forma de recuperarlo fue abrir un ticket de soporte
Si un consumidor de un slot de replicación deja de leer, necesariamente debería dispararse una alerta
Otra razón es que la ruta de código para obtener el snapshot inicial de una tabla y la ruta de código para leer cambios son completamente distintas. Inicializar la lectura de un slot de replicación de modo que no se pierda ni un solo cambio no es trivial
Lamentablemente, desde la perspectiva de captura de cambios, la replicación es la solución menos hacky
Yo uso polling, pero en lugar de
updated_atguardo el txidMe da curiosidad qué comportamiento preferirías
Si manejas grandes volúmenes de datos, vas a querer tratar el snapshot inicial y la lectura de cambios de forma diferente. Porque deberías poder hacer cosas como inicialización en paralelo o inicialización basada en backups físicos. Dicho eso, entiendo que podría ser útil una función que, después de crear el slot, transmita selectivamente los datos existentes
La parte de inicializar la lectura del slot de replicación para no perder cambios no debería ser difícil, me parece; me da curiosidad saber dónde te trabaste
Cuando no necesitas todos los cambios, también son útiles los slots de replicación temporales, que se limpian solos si se corta la conexión. También hay una configuración para definir un máximo de WAL retenido y no matar el servidor
Estaría bueno que explicaras más cómo usas txid en lugar de
updated_at