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. Faça login como um usuário com o privilégio CREATE no banco de dados e execute a seguinte consulta:

CREATE PUBLICATION <publication name>;

Copy

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

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

Copy

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

5. Como administrador de conta Snowflake, execute as seguintes tarefas:

   1. Crie um usuário Snowflake com o tipo SERVICE. Crie um banco de dados para armazenar os dados replicados e configure privilégios para que o usuário Snowflake crie objetos nesse banco de dados, concedendo os privilégios USAGE e CREATE SCHEMA.

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
                WAREHOUSE_SIZE = 'MEDIUM'
                AUTO_SUSPEND = 300
                AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

      Copy

   2. Crie um par de chaves seguras (pública e privada). Armazene a chave privada do usuário em um arquivo para fornecer à configuração do conector. Atribua a chave pública ao usuário do serviço Snowflake:

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      Copy

      Para obter mais informações, consulte par de chaves.

   3. Designar um warehouse para o conector usar. Comece com o tamanho do warehouse MEDIUM e, em seguida, experimente o tamanho, dependendo da quantidade de tabelas que estão sendo replicadas e da quantidade de dados transferidos. Os números de tabelas grandes normalmente são mais bem dimensionados com warehouses multicluster, em vez do tamanho do warehouse.

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.

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: <nome da tabela de origem>_JOURNAL_<carimbo de data/hora>_<geração do esquema>

Onde <carimbo de data/hora> é o valor de segundos de época em que a tabela de origem foi adicionada à replicação e <geração de esquema> é um número inteiro que aumenta a cada alteração de esquema na tabela de origem. Isso significa que uma tabela de origem que passa por alterações de esquema terá várias 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 só faz uso ativo do último diário para cada tabela de origem replicada. Se quiser recuperar o armazenamento, você pode eliminar com segurança as tabelas de diário relacionadas às tabelas de origem que foram removidas da replicação e todas as tabelas replicadas ativamente, exceto as de última geração.

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.