Configure o Openflow Connector for MySQL¶

Pré-requisitos¶

1. Certifique-se de ter revisado Sobre a Openflow Connector for MySQL.

2. Certifique-se de que você tenha um MySQL 8 ou posterior para sincronizar os dados com o Snowflake.

3. Certifique-se de ter configurado o Openflow.

4. Como administrador de banco de dados, execute as seguintes tarefas:

   1. Habilite os logs binários, salve e configure seu formato da seguinte forma:

      log_bin	Defina como on. Isso habilita o log binário que registra alterações estruturais e de dados.
      binlog_format	Defina como row. O conector suporta apenas a replicação baseada em linhas. Versões MySQL 8.x podem ser as últimas a suportar essa configuração, e as versões futuras suportarão apenas a replicação baseada em linhas. Não aplicável em GCP Cloud SQL, onde é fixado no valor correto.
      binlog_row_metadata	Defina como full. O conector requer todos os metadados da linha para operar, principalmente os nomes das colunas e as informações da chave primária.
      binlog_row_image	Defina como full. O conector exige que todas as colunas sejam gravadas no registro binário. Não aplicável no Amazon Aurora, onde é fixado no valor correto.
      binlog_row_value_options	Deixe em branco. Essa opção afeta apenas as colunas JSON, onde pode ser definida para incluir apenas as partes modificadas dos documentos JSON para as instruções UPDATE. O conector exige que os documentos completos sejam gravados no registro binário.
      binlog_expire_logs_seconds	Defina como pelo menos algumas horas, ou mais, para garantir que o agente de banco de dados possa continuar a replicação incremental após pausas prolongadas ou tempo de inatividade. A Snowflake recomenda que você defina o período de expiração do log binário (binlog_expire_logs_seconds) para pelo menos algumas horas para garantir o funcionamento estável do conector. Após o término do período de expiração do log binário, os arquivos de log binário podem ser removidos automaticamente. Se a integração for pausada por um longo período, por exemplo, devido a trabalhos de manutenção, e os arquivos de logs binários expirados forem excluídos durante esse período, o Openflow não poderá replicar os dados desses arquivos. Se você estiver usando a replicação programada, o valor precisa ser maior do que o cronograma configurado.

      Veja o código a seguir como exemplo:

      log_bin = on
      binlog_format = row
      binlog_row_metadata = full
      binlog_row_image = full
      binlog_row_value_options =
      

      Copy

   2. Aumente o valor de sort_buffer_size.

      sort_buffer_size = 4194304
      

      Copy

      sort_buffer_size define a quantidade de memória (em bytes) alocada por thread de consulta para operações de classificação na memória, como ORDER BY. Se o valor for muito pequeno, o conector poderá falhar com a seguinte mensagem de erro:

      Out of sort memory, consider increasing server sort buffer size. Isso indica que sort_buffer_size deve ser aumentado.

   3. Se você estiver usando bancos de dados do Amazon RDS, aumente o período de retenção relevante para binlog_expire_logs_seconds usando rds_set_configuration. Por exemplo, se você quiser armazenar o log binário por 24 horas, chame mysql.rds_set_configuration('binlog retention hours', 24).

   4. Conecte-se via SSL. Se estiver planejando usar uma conexão SSL para MySQL, prepare o certificado raiz para o servidor de banco de dados. Ele é necessário durante a configuração.

   5. Crie um usuário para o conector. O conector requer um usuário com os privilégios REPLICATION_SLAVE e REPLICATION_CLIENT para ler os logs binários. Conceda esses privilégios:

      GRANT REPLICATION SLAVE ON *.* TO '<username>'@'%'
      GRANT REPLICATION CLIENT ON *.* TO '<username>'@'%'
      

      Copy

   6. Conceda o privilégio SELECT em cada tabela replicada:

      GRANT SELECT ON <schema>.* TO '<username>'@'%'
      GRANT SELECT ON <schema>.<table> TO '<username>'@'%'
      

      Copy

      Para obter mais informações sobre a segurança da replicação, consulte Log binário.

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.

Configuração do conector¶

Como engenheiro de dados, execute as seguintes tarefas para instalar e configurar o conector:

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

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

Execute o fluxo¶

1. Clique com o botão direito do mouse no plano e selecione Enable all Controller Services.

2. Clique com o botão direito do mouse no grupo de processos importado e selecione Start. O conector inicia a ingestão de dados.