Configure o Openflow Connector for SQL Server¶

Parâmetros de fluxo¶

Comece definindo os parâmetros do contexto de parâmetros de origem do SQLServer e, em seguida, do contexto de parâmetros de destino do SQLServer. Depois de concluir esse procedimento, habilite o conector. O conector se conecta a ambos SQLServer e Snowflake e começa a ser executado. No entanto, o conector não replica nenhum dado até que as tabelas a serem replicadas sejam explicitamente adicionadas à sua configuração.

Para configurar tabelas específicas para replicação, edite o contexto Parâmetros de ingestão do SQLServer. Depois que você aplicar as alterações ao contexto Parâmetros de ingestão do SQLServer, a configuração será captada pelo conector e o ciclo de vida da replicação será iniciado para cada tabela.

Replicar tabelas de um servidor de réplica do SQL Server¶

O conector pode ingerir dados de um servidor primário, de um servidor de assinante usando uma replicação transacional ou de uma réplica secundária em um grupo de disponibilidade Always On. Antes de configurar o conector para se conectar a uma réplica do SQL Server, certifique-se de que a replicação entre os 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 e os eventos de rastreamento de alterações estejam presentes no servidor de réplica usado pelo conector.

Para garantir a continuidade, garanta que o mesmo usuário de conexão esteja disponível nos servidores tanto primário quanto de réplica e tenha acesso aos dados e às tabelas de rastreamento de alterações.

Grupos de disponibilidade Always On¶

Grupos de disponibilidade Always On são uma solução de alta disponibilidade e recuperação de desastres que mantém cópias sincronizadas de bancos de dados para fins de failover. O conector pode ler de uma réplica secundária no grupo de disponibilidade. Para obter a melhor experiência, configure um ouvinte de grupo de disponibilidade e use o nome DNS do ouvinte no parâmetro SQLServer Connection URL.

Reiniciar a replicação da tabela¶

Uma tabela no estado FAILED, por exemplo, devido a uma chave primária ausente ou alteração de esquema incompatível, não é reiniciada automaticamente. Se uma tabela entrar no estado FAILED, ou se você precisar reiniciar a replicação do zero, use o procedimento a seguir para remover e adicionar novamente a tabela à replicação.

1. Remova a tabela dos parâmetros de fluxo: no contexto de parâmetros de ingestão, remova a tabela de Included Table Names ou modifique o Included Table Regex para que a tabela não seja mais correspondida.

2. Verifique se a tabela foi removida:

   1. Na tela do tempo de execução do Openflow, clique com o botão direito do mouse em um grupo de processadores e escolha Controller Services.

   2. Na tabela com a lista de serviços do controlador, localize a linha Table State Store, clique nos três pontos verticais à direita da linha e escolha View State.

   Importante

   É necessário aguardar até que o estado da tabela seja totalmente removido da lista antes de prosseguir. Não continue até que esta alteração de configuração tenha sido concluída.

3. Limpe o destino: assim que o estado da tabela for exibido como totalmente removido, execute DROP manualmente da tabela de destino no Snowflake. Observe que o conector não substituirá uma tabela de destino existente durante a fase do instantâneo. Se a tabela ainda existir, a replicação falhará novamente. Opcionalmente, a tabela de diário e o fluxo também podem ser removidos se não forem mais necessários.

4. Adicione a tabela novamente: atualize os parâmetros Included Table Names ou Included Table Regex para incluir a tabela novamente.

5. Verifique a reinicialização: consulte Table State Store seguindo as instruções já apresentadas. O estado da tabela deve aparecer com status NEW, depois passar para SNAPSHOT_REPLICATION e, por fim, INCREMENTAL_REPLICATION.

Replique um subconjunto de colunas em uma tabela¶

O conector filtra 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.

Inclua ou exclua colunas 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. Os campos schema e table são obrigatórios. É necessário um ou mais de included, excluded, includedPattern e excludedPattern.

[
    {
        "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 o estado atual dos dados das tabelas de origem e o estado das linhas 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: <source_table_name>_JOURNAL_<timestamp>_<schema_generation>, em que <timestamp> é o valor de segundos da época quando a tabela de origem foi adicionada à replicação e <schema_generation> é 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 você remove uma tabela da replicação e depois a adiciona de volta, o valor <timestamp> muda e <schema_generation> começa novamente a partir de 1.

O conector nunca descarta tabelas de diário, mas usa o diário mais recente para cada tabela de origem replicada, lendo apenas os fluxos somente para anexação acima dos 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.

Use a expressão CRON no parâmetro CRON do cronograma da tarefa de mesclagem para limitar o custo do warehouse e as mesclagens somente ao horário agendado. Ele controla o fluxo dos MergeSnowflakeJournalTable que chegam ao processador e as mesclagens são acionadas somente em um período de tempo específico. Para obter mais informações sobre agendamento, consulte <https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#scheduling-strategy>Estratégia de agendamento _.