Configure o Openflow Connector for PostgreSQL¶

Configuração de wal_level¶

Openflow Connector for PostgreSQL requer que wal_level seja definido como logical.

Dependendo de onde o servidor PostgreSQL estiver hospedado, você pode configurar o wal_level da seguinte forma:

Crie uma publicação¶

O Openflow Connector for PostgreSQL requer que uma publicação seja criada e configurada em PostgreSQL antes do início da replicação. Você pode criá-la para todas as tabelas ou para um subconjunto delas, bem como para tabelas específicas apenas com colunas especificadas. Certifique-se de que todas as tabelas e colunas que você planeja replicar estejam incluídas na publicação. Você também pode modificar a publicação posteriormente, enquanto o conector estiver em execução. Para criar e configurar uma publicação, faça o seguinte:

1. Efetue login como um usuário com o privilégio CREATE no banco de dados e execute a seguinte consulta:

   CREATE PUBLICATION <publication name>; 

   Copy

2. Defina as tabelas que o agente do banco de dados poderá ver usando:

ALTER PUBLICATION <publication name> ADD TABLE <table name>; 

Copy

Importante

PostgreSQL 15 e posteriores oferecem suporte à configuração de publicações para um subconjunto específico de colunas da tabela. Para que o conector seja corretamente compatível com isso, você deve usar as configurações de filtragem de coluna para incluir as mesmas colunas definidas na publicação.

Sem essa configuração, o conector se comportará da seguinte forma:

• Na tabela de destino, as colunas que não estiverem incluídas no filtro serão sufixadas com __DELETED. Todos os dados replicados durante a fase de instantâneo serão mantidos.

• Após adicionar novas colunas à publicação, a tabela falhará permanentemente e você precisará reiniciar a replicação.

Para obter mais informações, consulte ALTER PUBLICATION.

Replique um conjunto de tabelas em tempo real¶

1. Clique com o botão direito do mouse no grupo de processos importado e selecione Parameters.

2. Preencha os valores dos parâmetros necessários conforme descrito em Parâmetros de fluxo.

Replicação de tabelas de um servidor de replicação PostgreSQL¶

O conector pode ingerir dados de um servidor primário, uma réplica de espera ativa ou servidor de assinante usando replicação lógica. Antes de configurar o conector para se conectar a uma réplica do PostgreSQL, certifique-se de que a replicação entre nós primários e de réplica funcione corretamente. Ao investigar problemas de dados ausentes no conector, primeiro certifique-se de que as linhas ausentes estejam presentes no servidor de réplica usado pelo conector.

Considerações adicionais ao conectar-se a uma réplica em espera:

• Somente a conexão a réplicas em espera ativa é compatível. Observe que as réplicas em espera dinâmica não podem aceitar conexões de clientes até serem promovidas a uma instância primária.

• A versão do PostgreSQL do servidor deve ser >= 16.

• A publicação necessária para o conector deve ser criada no servidor primário, não no servidor em espera. O servidor em espera é somente leitura e não permite criar publicação.

Se você se conectar a uma instância de espera ativa e consulte o erro Tentativa de criar o slot de replicação “<replication slot> esgotou o tempo limite. Se estiver conectando a uma instância em espera, certifique-se de que haja algum tráfego na instância PostgreSQL primária, caso contrário a chamada para criar um slot de replicação nunca retornará. no código do Openflow ou o processador Read PostgreSQL CDC Stream não está iniciando, faça login na instância PostgreSQL primária e execute a seguinte consulta:

SELECT pg_log_standby_snapshot(); 

O erro ocorre quando não há alterações de dados no servidor primário. Como tal, o conector pode travar enquanto cria um slot de replicação no servidor de réplica. Isso resulta no servidor de réplica que exige informações sobre a execução de transações do servidor primário para poder criar um slot de replicação. Os servidores primários não enviarão as informações enquanto estiverem ociosos. A função pg_log_standby_snapshot() força o servidor primário a enviar informações sobre transações em execução para o servidor de réplica.

Remova e adicione novamente uma tabela à replicação¶

Para remover uma tabela da replicação, certifique-se de que ela seja removida dos parâmetros Nomes de tabela inclusos ou Regex de tabela inclusa no contexto Parâmetros de replicação.

Se quiser adicionar novamente a tabela à replicação mais tarde, primeiro exclua a tabela de destino correspondente no Snowflake. Após isso, adicione a tabela novamente aos parâmetros Nomes de tabela inclusos ou Regex de tabela inclusa. Isso garante que o processo de replicação comece do zero para a tabela.

Essa abordagem também pode ser usada para se recuperar de um cenário de replicação de tabela com falha.

Replique um subconjunto de colunas em uma tabela¶

O conector pode filtrar os dados replicados por tabela para um subconjunto de colunas configuradas.

Para aplicar filtros a colunas, modifique a propriedade Filtro de coluna no contexto Parâmetros de replicação, adicionando uma matriz de configurações – uma entrada para cada tabela à qual deseja aplicar um filtro.

As colunas podem ser incluídas ou excluídas por nome ou padrão. Você pode aplicar uma única condição por tabela ou combinar várias condições, com as exclusões sempre tendo precedência sobre as inclusões.

O exemplo a seguir mostra os campos disponíveis. schema e table são obrigatórios e, em seguida, um ou mais dos campos included, excluded, includedPattern, excludedPattern são obrigatórios.

[  {  "schema": "<source table schema>",  "table" : "<source table name>",  "included": ["<column name>", "<column name>"],  "excluded": ["<column name>", "<column name>"],  "includedPattern": "<regular expression>",  "excludedPattern": "<regular expression>",  } ] 

Monitore as alterações de dados em tabelas¶

O conector replica não apenas o estado atual dos dados das tabelas de origem, mas também cada estado de cada linha de cada conjunto de alterações. Esses dados são armazenados em tabelas de diário criadas no mesmo esquema da tabela de destino.

Os nomes das tabelas de diário são formatados como: é um número inteiro que aumenta a cada alteração de esquema na tabela de origem. Como resultado, as tabelas de origem que passam por alterações de esquema terão vários tabelas de diário.

Quando uma tabela é removida da replicação e depois adicionada novamente, o valor de <carimbo de data/hora> será alterado e a <geração de esquema> começará novamente a partir de 1.

O conector nunca descarta tabelas de diário, mas faz uso do diário mais recente para cada tabela de origem replicada, lendo apenas fluxos apenas para anexação sobre os diários. Para recuperar o armazenamento, você pode:

• Truncar todas as tabelas de diário a qualquer momento.

• Descartar as tabelas de diário relacionadas às tabelas de origem que foram removidas da replicação.

• Descartar todas as tabelas de diário, exceto as de última geração, para tabelas replicadas ativamente.

Por exemplo, se o conector estiver definido para replicar ativamente a tabela de origem orders e você tiver removido anteriormente a tabela customers da replicação, poderá ter as seguintes tabelas de diário. Nesse caso, você pode descartar todas elas exceto orders_5678_2.

customers_1234_1 customers_1234_2 orders_5678_1 orders_5678_2 

Configure o agendamento de tarefas de fusão¶

O conector usa um warehouse para mesclar informações de captura de dados de alteração (CDC) nas tabelas de destino. Essa operação é acionada pelo processador MergeSnowflakeJournalTable. Se não houver novas alterações ou se não houver novos FlowFiles aguardando na fila MergeSnowflakeJournalTable, nenhuma fusão será acionada e o warehouse será suspenso automaticamente.

Para limitar o custo do warehouse e limitar as mesclagens apenas ao horário programado, use a expressão CRON no parâmetro CRON Cronograma da tarefa de mesclagem. Ele controla o fluxo dos FlowFiles que chegam ao processador MergeSnowflakeJournalTable e as mesclagens são acionadas somente em um período de tempo específico. Para obter mais informações sobre agendamento, consulte Estratégia de agendamento.

Pare ou exclua o conector¶

Ao interromper ou remover o conector, você deve considerar o slot de replicação que o conector usa.

O conector cria seu próprio slot de replicação com um nome que começa com snowflake_connector_ seguido de um sufixo aleatório. À medida que o conector lê o fluxo de replicação, ele avança o slot, de modo que o PostgreSQL possa cortar o log WAL e liberar espaço em disco.

Quando o conector é pausado, o slot não é avançado e as alterações no banco de dados de origem continuam aumentando o tamanho do registro WAL. Você não deve manter o conector em pausa por longos períodos de tempo, especialmente em bancos de dados de alto tráfego.

Quando o conector é removido, seja excluindo-o da tela do Openflow ou por qualquer outro meio, como a exclusão de toda a instância do Openflow, o slot de replicação permanece no lugar e deve ser removido manualmente.

Se você tiver várias instâncias do conector replicando do mesmo banco de dados PostgreSQL, cada instância criará seu próprio slot de replicação com nome exclusivo. Ao descartar um slot de replicação manualmente, certifique-se de que seja o correto. Você pode ver qual slot de replicação é usado por uma determinada instância do conector verificando o estado do processador CaptureChangePostgreSQL.