Todas as formas de capturar mudanças no Postgres
(blog.sequin.io)- Para enviar mudanças do Postgres em tempo real para outros sistemas, é necessário CDC (Change Data Capture), e entre opções que vão de notificações simples à replicação baseada em WAL, a confiabilidade e a carga operacional variam bastante
- Listen/Notify é a forma mais leve de começar, mas por causa da semântica de entrega at-most-once, das notificações efêmeras e do limite de payload de 8000 bytes, fica mais próximo de um sinal auxiliar do que de um CDC central
- Polling de tabelas e tabelas de auditoria (outbox pattern) podem ser implementados apenas com tabelas e gatilhos padrão, mas exigem resolver manualmente detecção de exclusão, diff, ordem de commit, write amplification e backpressure
- Replicação lógica (logical replication) é uma forma poderosa de fazer streaming de insert/update/delete a partir do WAL, mas a aplicação precisa gerenciar replication slots, ack, reinicialização e adaptação à vazão
- O Sequin, com base na replicação lógica do Postgres, envia mudanças para SQS, Kafka, Elasticsearch, Redis, endpoints HTTP e outros destinos, reduzindo a necessidade de lidar diretamente com replication slots
Quando o Postgres CDC é necessário
- O Postgres é forte para lidar com dados armazenados, mas se você quiser acionar workflows com mudanças em tabelas ou fazer streaming em tempo real para outros armazenamentos de dados, sistemas ou serviços, será preciso projetar separadamente a movimentação de dados
- Change Data Capture (CDC) é a abordagem de identificar e capturar mudanças no banco de dados e depois entregá-las em tempo real para sistemas downstream
- Há várias formas de capturar mudanças no Postgres, e elas diferem em dificuldade de implementação, confiabilidade e carga operacional
Listen/Notify: o pub-sub mais simples
- O Listen/Notify do Postgres é um recurso de comunicação entre processos que funciona no padrão publish-subscribe
- Uma sessão pode dar
listenem um canal específico, e atividades no banco de dados ou outras sessões podem enviarnotifypara esse canal - Para captura de mudanças, ele pode ser usado com gatilhos
- Um gatilho de exemplo cria um JSON com
table,ideactiondo registro alterado no momento deafter insert or update or delete, e chamapg_notify('table_changes', payload::text)
- Um gatilho de exemplo cria um JSON com
- As limitações são bem claras
- Tem semântica de entrega at-most-once, e o listener precisa estar conectado no momento em que a notificação é publicada
- O listener só recebe notificações a partir do momento da assinatura, então mesmo uma desconexão temporária por problema de rede pode fazer notificações serem perdidas
- O limite de tamanho do payload é de 8000 bytes, e se isso for excedido o comando
notifyfalha - O tamanho do payload inclui também o nome do canal, e, como um identificador do Postgres, o nome do canal pode ter no máximo 64 bytes
- Pode servir para detecção básica de mudanças ou otimização de polling de tabelas, mas pode não se encaixar bem em requisitos de CDC mais complexos
Polling de tabelas: simples, mas fraco para exclusões e diff
- A forma mais simples e robusta de capturar mudanças é fazer polling diretamente nas tabelas
- Cada tabela precisa de uma coluna como
updated_at, atualizada sempre que uma linha é modificada, e isso pode ser feito com gatilhos, se necessário - A combinação de
updated_atcomidpode ser usada como cursor, e a lógica da aplicação armazena e gerencia esse cursor - Se você usar também uma assinatura de Notify, a aplicação pode ser avisada sobre inserções e alterações de registros, reduzindo a frequência do polling
- Como as notificações do Postgres são efêmeras, faz mais sentido usá-las apenas como otimização sobre o polling
- Há três desvantagens principais
- Como linhas excluídas não permanecem na tabela, a detecção de exclusão é impossível
- Como alternativa, um gatilho de delete pode gravar
ide colunas necessárias em uma tabela separada, comodeleted_contacts, e a aplicação pode fazer polling dessa tabela - Você consegue saber que um registro foi atualizado, mas não o que mudou
- Como datetime e sequence do Postgres podem ter ordem diferente da ordem de commit, ao ler um bloco com base em
updated_atvocê pode deixar passar linhas que ainda estavam em commit
- Para rastreamento simples de mudanças, em que exclusões, diff e perdas ocasionais não são grandes problemas, é uma opção razoável
Tabelas de auditoria: armazenando um log de mudanças com outbox pattern
- A abordagem de tabela de auditoria (audit table) registra mudanças em uma tabela separada
changelog, também chamada de outbox pattern - A
changelogpode ter colunas relacionadas à mudançaaction: se foiinsert,updateoudeleteold: ojsonbdo registro antes da mudança, vazio em insertsvalues: ojsonbdos campos alterados, vazio em deletesinserted_at: o momento em que a mudança ocorreu
- Para implementar isso, você precisa de uma função de gatilho que insira na
changelogsempre que houver uma mudança, além de gatilhos para cada tabela monitorada - Também é possível consumir a
changelogcomo se fosse uma fila- Um worker da aplicação busca mudanças na tabela
- Para um processamento aproximadamente exactly-once, pode-se usar
for update skip lockeddo Postgres - O worker pode abrir uma transação, bloquear um lote com
order by timestamp limit 100 for update skip locked, processá-lo, apagar os registros processados e depois fazer commit
- Há desvantagens operacionais
- Uma única escrita em tabela gera write amplification, criando várias escritas na tabela de auditoria
- Em geral, há pelo menos três escritas: insert inicial na tabela de auditoria, update durante o processamento e delete após o processamento
- O modelo de fan-out via workers precisa ser desenhado manualmente para a sua aplicação
- Antes de um deploy em escala de produção, é provável que seja necessário ajustar a função de gatilho e o desenho das tabelas
- Também vale considerar políticas detalhadas, como limite de tempo que um worker pode manter uma mudança em checkout
- Mesmo que o worker não processe com sucesso, a tabela de auditoria continua enchendo, então o gerenciamento de backpressure é insuficiente
Foreign Data Wrapper: uma opção mais próxima de sincronização entre Postgres específicos
- Foreign Data Wrapper (FDW) é um recurso que permite a um banco Postgres ler e escrever em fontes de dados externas
- A extensão baseada em FDW com suporte mais amplo é
postgres_fdw- Ela permite conectar dois bancos Postgres e criar, em um deles, uma estrutura parecida com uma view que referencia tabelas do outro
- Internamente, um banco Postgres atua como cliente e o outro como servidor
- Ao consultar uma foreign table, o banco cliente envia consultas ao banco servidor via o wire protocol do Postgres
- FDW não é comum como abordagem de captura de mudanças, e fora de situações muito específicas dificilmente é recomendável
- Se o objetivo for escrever mudanças de um banco Postgres em outro banco Postgres, o FDW pode fazer sentido
- Um exemplo é o uso de um banco para contabilidade e outro para a aplicação
- Em vez de passar por uma etapa intermediária de captura de mudanças, é possível refletir diretamente entre bancos com
postgres_fdw
- Também é possível criar um FDW próprio para fazer POST de mudanças para uma API interna
- Como a escrita na API acontece dentro do commit, a API pode rejeitar a mudança e provocar rollback do commit
- O FDW é poderoso, mas raramente é a melhor opção para CDC, e escrever um FDW próprio é uma das maiores empreitadas entre as abordagens de captura de mudanças
- Criar um FDW próprio ficou mais fácil com ferramentas como Supabase wrappers, mas ainda assim é um trabalho grande
Replicação lógica direta: CDC poderoso com base em WAL
- O Postgres tem um protocolo para replicação de banco de dados, e um de seus modos é a replicação lógica (logical replication)
- A replicação lógica é construída sobre o WAL (write-ahead log) do Postgres
- Todo insert, update e delete do banco é rastreado
- As mudanças são transmitidas em streaming para o subscriber
- O usuário primeiro cria um replication slot no primary
- Usa-se a forma
pg_create_logical_replication_slot('<your_slot_name>', '<output_plugin>')
- Usa-se a forma
- O
output_plugindefine qual plugin fará o decode das mudanças do WALpgoutputé o plugin padrão e produz saída no formato binário esperado pelo servidor clientetest_decodingé um plugin de saída simples que fornece mudanças do WAL em formato legível por humanos- Embora não seja um plugin embutido no Postgres,
wal2jsoné um plugin popular, e JSON costuma ser um ponto de partida mais fácil para aplicações do que o formato binário do Postgres
- Depois de criar o replication slot, é possível iniciar e consumir
- O replication slot usa uma parte do protocolo do Postgres diferente da usada por consultas padrão
- Várias bibliotecas cliente oferecem funções para ajudar a trabalhar com replication slots
- Em
psycopg2, por exemplo, é possível consumir mensagens do WAL comcursor.start_replication(...)ecursor.consume_stream(...), e enviar ack comcursor.send_feedback(flush_lsn=msg.wal_end)
- O cliente precisa enviar ack das mensagens do WAL recebidas, e o replication slot funciona de forma parecida com um Kafka com offset
- A replicação lógica é uma abordagem robusta criada para CDC, mas é complexa
- Replication slots e o replication protocol são menos familiares para desenvolvedores do que tabelas e consultas normais
- É necessário ter uma estratégia para não perder mensagens durante reinicializações
- Também é preciso projetar o sistema para processar o grande volume de mensagens que pode sair do Postgres
Sequin: uma ferramenta de CDC sobre replicação lógica
- Sequin é uma ferramenta de CDC que envia mudanças e linhas do Postgres para filas, streams, índices de busca, caches, endpoints HTTP e outros destinos
- Entre os destinos estão SQS, Kafka, Elasticsearch, Redis e endpoints HTTP
- O Sequin usa internamente a replicação lógica do Postgres, mas abstrai a complexidade do protocolo de baixo nível
- Ele consegue capturar insert, update e delete, e em updates e deletes captura tanto os valores
newquantooldda linha - Faz sentido considerar o Sequin quando
- você precisa de CDC em tempo real
- quer fazer streaming direto para destinos como SQS ou webhook, sem um sistema intermediário
- precisa de recursos como backfill de dados históricos e filtragem de mudanças com cláusulas SQL
where - quer uma alternativa mais simples do que gerenciar replication slots diretamente
- precisa de garantia de processamento exactly-once
- Também há desvantagens
- O Sequin não é uma extensão interna do Postgres, mas uma ferramenta de terceiros executada ao lado do banco de dados
- Por não ser uma extensão, tem ampla compatibilidade com praticamente qualquer banco Postgres, mas, se você não usar o Sequin Cloud, terá de montar a infraestrutura adicional por conta própria
Critérios de escolha
- No começo, Listen/Notify e polling de tabelas são adequados
- Listen/Notify é bom para capturar eventos não críticos, prototipagem e otimização de polling
- Polling é uma solução direta e confiável para casos de uso simples
- Em um estágio um pouco mais sério, tabelas de auditoria podem ser uma opção intermediária
- Elas podem capturar payloads
neweolddas linhas - Se bem implementadas, podem oferecer um sistema de processamento exactly-once
- Em escala, write amplification e ausência de backpressure viram problema, e erros na configuração manual podem fazer mensagens se perderem
- Elas podem capturar payloads
- Em escala maior, a replicação lógica é a opção mais próxima de uma solução robusta
- Ainda assim, é recomendado usar uma ferramenta como o Sequin em vez de ler diretamente do slot
- FDW é um recurso interessante, mas tem baixa chance de resolver necessidades gerais de CDC
1 comentários
Opiniões no Hacker News
Triggers + tabelas de histórico (tabelas de auditoria) são a resposta certa em 98% dos casos. Se você ainda não usa, pode começar hoje. É uma técnica comprovada há mais de 30 anos.
Um exemplo simples de implementação genérica está em https://gist.github.com/slotrans/353952c4f383596e6fe8777db5d.... É uma abordagem que abre mão de eficiência de espaço em favor de uma “implementação fácil”.
Seria ótimo se fosse possível armazenar dados imutáveis, mas provavelmente há uma quantidade enorme de dados mutáveis no seu banco de dados, e você provavelmente esquece muitas coisas todos os dias. Não se esqueça: use tabelas de histórico.
Referência: https://github.com/matthiasn/talk-transcripts/blob/master/Hi...
É melhor não usar bibliotecas ou técnicas de rastreamento de histórico na camada de aplicação, como Papertrail. São lentas, propensas a erros e não capturam alterações no DB que contornam a stack da aplicação. A tentativa de definir timestamps de
updatedpela aplicação também é fundamentalmente equivocada, porque cada servidor web tem um relógio diferente. É preciso usar o relógio do DB, que é o único relógio correto.now()dentro da query.Mas sincronizar apenas com esse timestamp não é suficiente. Isso porque o timestamp é gerado no momento de início da transação, não no momento do commit da transação.
Se você fizer polling da tabela filtrando pelo timestamp mais recente, pode perder parte de transações cuja ordem de commit ficou misturada. Dá para criar uma janela de segurança consultando alguns minutos mais para trás e removendo duplicatas, mas no PostgreSQL o tempo de uma transação é ilimitado, e consultar um passado distante demais gera muito desperdício. Se precisão e eficiência importam, essa abordagem não é adequada.
Com número de sequência do log, horário do DB e
REPLICA IDENTITY FULL, ele inclui até os estados antes/depois das mudanças. Depois, ao materializar coleções em lugares como o Snowflake, você obtém por padrão tabelas sincronizadas que acompanham as atualizações do DB de origem.Também é possível transformar ou materializar o histórico completo das tabelas para fins de auditoria a partir do mesmo data lake subjacente, sem precisar anexar novamente captura ou um leitor de WAL ao DB de origem.
Também descrevi separadamente uma abordagem em SQLite que implementa um padrão parecido usando colunas em vez de JSON: https://simonwillison.net/2023/Apr/15/sqlite-history/
Este texto resume bem, de forma concisa, várias abordagens possíveis com recursos nativos do Postgres.
Na parte de “captura de alterações em tabelas de auditoria”, usei bem o padrão de Temporal Tables em uma empresa anterior. Diferentemente de outros grandes SGBDs relacionais, o Postgres não traz isso embutido, mas há um padrão simples que pode ser usado por meio de funções SQL: https://github.com/nearform/temporal_tables
Ele permite ver o estado de uma tabela em um ponto específico no tempo, respondendo a perguntas como “quais eram as configurações deste usuário em 12 de agosto?”, “quantos registros pendentes havia ontem à noite às 23:55?” e “mostre a diferença entre as flags de recurso atuais e as de uma semana atrás”.
Certa vez prestei consultoria para uma empresa que tinha um SQL Server monolítico enorme. Não era Postgres, mas seria parecido se fosse Postgres.
Ele estava em operação havia décadas e era usado para todo tipo de finalidade dentro da empresa; na prática, todos os aplicativos e processos de negócio da empresa inteira armazenavam dados nesse banco.
O problema era que havia muitos aplicativos consultando esse DB, além de uma quantidade enorme de processos e procedimentos que inseriam e modificavam dados, então mudanças ou acréscimos nos processos upstream de inserção/modificação acabavam quebrando invariantes no nível da aplicação. Mesmo processos normais se comportavam de forma diferente quando havia dados ruins.
Era muito difícil rastrear a causa, porque aquilo que se investigava geralmente tinha sido escrito 10 anos antes, e esses funcionários já tinham saído da empresa.
Fico me perguntando se seria possível capturar mudanças em um banco Postgres na forma de algum DAG, para saber quais processos inserem, modificam e removem dados e como historicamente se comportaram, além de como vários aplicativos consultam esses dados e como as estatísticas de consulta mudam ao longo do tempo.
Não sei bem se há precedentes para isso nem qual abordagem permitiria criar uma ferramenta assim. Já pensei em criar algo parecido no passado, mas parece uma área que exige um entendimento no nível de engenheiro do core do Postgres para fazer boas escolhas.
Você não obtém, para cada mudança, dados de origem no nível do cliente.
Ainda assim, há como contornar isso. O stream de replicação lógica também pode incluir mensagens informativas da função
pg_logical_emit_message, então o cliente pode inserir metadados diretamente. Talvez dê para configurar o cliente para emitir um identificador no início de cada transação.last updated by). Pode ser um antipadrão, então seria bom ter uma solução mais robusta.Essa abordagem funciona em praticamente todas as soluções da família SQL que usam WAL ou triggers.
Já usei várias vezes a abordagem com triggers no SQL Server, mas registrar todas as consultas tende a deixar tudo mais lento. Projetar um mecanismo de inserção que não atrapalhe a operação não é perfeito, e talvez seja necessário usar amostragem.
Se você seguir o caminho de “tabelas de auditoria”, basta usar o pgaudit. É uma extensão comprovada em produção e, se você usa AWS, também está disponível no 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
Não precisa necessariamente fazer isso. Querer isso significa transformar as relações do Postgres em contratos. Nenhum serviço poderá persistir estado interno.
Pode ser possível se você estiver realmente comprometido com design orientado a domínio, mas é melhor usar um sistema baseado em eventos que seja leve e prático.
Alguma coisa baseada em eventos é 1000 vezes mais complexa.
A abordagem de fazer polling de uma coluna
updated_atnão é robusta em sua forma mais simples. Não há garantia de que as transações serão commitadas nessa ordem.updated_atda Row 1 é definido como2023-09-22 12:00:01.Pouco depois, a transação B começa, o
updated_atda Row 2 é definido como2023-09-22 12:00:02, e B commita primeiro.A consulta de polling é executada, vê a Row 2 como a mudança mais recente e atualiza o cursor para
2023-09-22 12:00:02; depois, quando A commita, a Row 1 acaba sendo perdida.Uma forma simples de evitar esse problema é não fazer polling quase em tempo real. A ordem acaba ficando consistente.
Uma sugestão mais robusta talvez seja usar uma sequence. Por exemplo, ter uma coluna
updated_at_idxque é incrementada a cada alteração de linha.Fico curioso se, usando um trigger before que insere
now(), os timestampsupdated_atdas duas linhas ainda podem ficar diferentes da ordem de commit das transações.updated_ate o timestamp de commit não precisam ser iguais, masupdated_atdeveria representar com precisão, em milissegundos ou microssegundos, a ordem dos commits._txid, que um trigger define como o ID da transação atual, em vez deupdated_at. Depois, ao fazer polling, usotxid_current()para verificar quais transações foram commitadas e quais ainda não foram.É um pouco delicado e muito fácil cometer erro de borda, mas está rodando bem em produção há alguns anos.
O texto é excelente.
Se você usa Elixir e Postgres, criei uma pequena biblioteca com uma abordagem semelhante para escutar mudanças no WAL: https://github.com/cpursley/walex
Esses métodos são todos meio ruins e, pessoalmente, acho que polling é o mais prático
Seria bom se o Postgres inovasse nessa área
Até isso entrar no padrão SQL, acho difícil surgir tração no espaço de kernel dos SGBDs relacionais. As opções são muitas e complexas, e as soluções bem-sucedidas no espaço do usuário também não costumam impor um ônus excessivo em termos de desempenho
A propósito, quem pesquisa essa área em geral tende para a abordagem de tabelas de auditoria. Isso porque ela mantém propriedades ACID consistentes dentro do banco de dados e mantém o Postgres como ponto único de falha, em vez de acrescentar proxies ou jobs de polling
Há uma grande lacuna no mundo dos dados. Em vez de perguntar ao armazenamento de dados pelo resultado, seria bom se os resultados das consultas fossem enviados incrementalmente por push
Fazemos muita análise em tempo real e streaming; dá para fazer processamento de streams e também tratar parte disso dentro do armazenamento de dados com views materializadas. Mas, depois que os dados entram no DB ou no data lake, para ver mudanças a jusante, na prática voltamos ao polling
Há poucas soluções limpas para reagir quando alguma situação acontece nos dados ou para atualizar a tela sem recarregar a página. As soluções deste texto também parecem mais gambiarras do que recursos de primeira classe
Se você quiser criar relatórios atualizados em tempo real sem recarregar a página, normalmente o caminho acaba sendo carregar os dados do DB e então enviar as mudanças para a GUI via Kafka e WebSocket. Aí você passa a operar uma arquitetura lambda estranha, em que parte da análise fica no código e parte no DB
Há inovação nessa área. KSQL e Kafka Streams conseguem emitir mudanças, o Materialize tem assinaturas, e o ClickHouse tem live views. Mas muitos recursos são novos ou estão em fase de preview e não se encaixam exatamente. Usei todos eles, mas senti que jogam trabalho demais para o desenvolvedor
Seria bom ter uma biblioteca que permitisse receber diretamente um feed de mudanças com uma opção como
[select * from orders with suscribe]. É uma área importante o suficiente, mas que recebeu pouca atençãoHá uma grande armadilha da replicação que o texto não aborda, e por isso eu não uso replicação
O Postgres tenta oferecer uma garantia muito forte de que os consumidores de slots de replicação não percam dados. Então, se um consumidor não consome os dados do slot, o Postgres gentilmente continua guardando os dados perdidos e, no fim, vai até o disco encher e o DB cair. Passei por isso durante prototipagem em dois DBs SaaS diferentes, e a única forma de recuperar foi abrir um chamado de suporte
Se um consumidor de slot de replicação parar de ler, um alerta precisa necessariamente disparar
Outro motivo é que o caminho de código para obter o snapshot inicial de uma tabela e o caminho de código para ler mudanças são completamente diferentes. Inicializar a leitura do slot de replicação de modo a não perder nenhuma mudança não é trivial
Infelizmente, do ponto de vista de captura de mudanças, a replicação é a solução menos gambiarra
Eu uso polling, mas armazeno txid em vez de
updated_atFico curioso sobre qual comportamento você preferiria
Se você lida com grandes volumes de dados, vai querer tratar o snapshot inicial e a leitura de mudanças de formas diferentes. Isso porque tarefas como inicialização paralela ou inicialização baseada em backup físico precisam ser possíveis. Ainda assim, entendo que um recurso que fizesse streaming seletivo dos dados existentes após a criação do slot poderia ser útil
A parte de inicializar a leitura do slot de replicação para não perder mudanças não deveria ser difícil; fico curioso para saber onde você travou
Quando você não precisa de todas as mudanças, slots de replicação temporários que se limpam sozinhos quando a conexão cai também são úteis. Também há uma configuração para definir o máximo de WAL retido, para não derrubar o servidor
Seria bom se você explicasse melhor como usa txid em vez de
updated_at