ApsaraMQ for Kafka:Criar um conector de origem MySQL

Como funciona

Ao criar um conector de origem MySQL pelo console do ApsaraMQ for Kafka, o sistema executa automaticamente as seguintes ações:

1. Ativa o DataWorks Basic Edition (gratuito).

2. Cria um workspace do DataWorks e um grupo de recursos exclusivo para integração de dados.

3. Gera tópicos de destino na sua instância do Kafka — um tópico por tabela de origem.

O grupo de recursos exclusivo (4 vCPUs, 8 GB de memória) executa as tarefas de sincronização de dados. Esses grupos usam assinatura mensal e se renovam automaticamente ao expirar.

<topic-prefix>_<source-table-name>

Pré-requisitos

Antes de começar, verifique se você tem:

• Uma instância do ApsaraMQ for Kafka com o recurso de conector habilitado, implantada em uma destas regiões: China (Shenzhen), China (Chengdu), China (Beijing), China (Zhangjiakou), China (Hangzhou), China (Shanghai) ou Singapura

• Uma instância do ApsaraDB RDS for MySQL com banco de dados, conta de banco de dados e pelo menos uma tabela criada. Para referência de SQL, consulte Instruções SQL comuns para MySQL

• Blocos CIDR não sobrepostos entre a VPC da sua instância do RDS MySQL e a VPC da sua instância do Kafka

• Autorização concedida ao DataWorks para acessar suas ENIs por meio da página de Autorização de Acesso a Recursos da Nuvem

• Tanto a fonte de dados (instância do ApsaraDB RDS for MySQL) quanto o destino dos dados (instância do ApsaraMQ for Kafka) criados na mesma conta Alibaba Cloud

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '<your-username>'@'%';

Criar e implantar o conector

1. Faça login no console do ApsaraMQ for Kafka.

2. Na seção Resource Distribution da página Overview, selecione a região da sua instância.

3. No painel de navegação à esquerda, clique em Connectors.

4. Na página Connectors, selecione sua instância na lista suspensa Select Instance e clique em Create Connector.

5. Conclua o assistente de três etapas: Clique em Next. Step 2: Configure source service Selecione ApsaraDB RDS for MySQL como serviço de origem e configure os parâmetros a seguir: Clique em Next. Step 3: Configure destination service Confirme a instância do Kafka de destino e clique em Create.

   Step 1: Configure basic information

   Parâmetro	Descrição	Exemplo
   Name	Nome exclusivo do conector dentro da instância. De 1 a 48 caracteres; permite dígitos, letras minúsculas e hifens (-). Não pode começar com hífen. O sistema cria automaticamente um consumer group chamado connect-<connector-name>.	kafka-source-mysql
   Instance	Exibe o nome e o ID da instância do Kafka selecionada.	demo alikafka_post-cn-st21p8vj****

   Parâmetro	Descrição	Exemplo
   Region of ApsaraDB RDS for MySQL Instance	Região da instância do RDS MySQL de origem.	China (Shenzhen)
   ApsaraDB RDS for MySQL Instance ID	ID da instância do banco de dados de origem.	rm-wz91w3vk6owmz****
   Database Name	Nome do banco de dados a ser sincronizado.	mysql-to-kafka
   Database Account	Nome de usuário para a conexão com o banco de dados.	mysql_to_kafka
   Password of Database Account	Senha para a conexão com o banco de dados.	--
   Database Table	Um ou mais nomes de tabelas, separados por vírgulas (,). Cada tabela corresponde a um tópico.	mysql_tbl
   Tables to Automatically Add	Expressão regular para detectar e sincronizar novas tabelas automaticamente. Use .* para todas as tabelas.	.*
   Topic Prefix	Prefixo para nomes de tópicos gerados automaticamente. Deve ser globalmente único.	mysql

6. Na página Connectors, localize o conector e clique em Deploy na coluna Actions. Quando a coluna Status exibir Running, o conector estará ativo e sincronizando dados.

Verificar a sincronização de dados

1. Insira uma linha de teste em uma tabela de origem: Para mais exemplos de SQL, consulte Instruções SQL comuns para MySQL.

      INSERT INTO mysql_tbl
          (mysql_title, mysql_author, submission_date)
          VALUES
          ("mysql2kafka", "tester", NOW());

2. Use o recurso de consulta de mensagens no ApsaraMQ for Kafka para confirmar que os dados chegaram ao tópico correspondente. Para instruções, consulte Consultar mensagens. Um evento INSERT sincronizado com sucesso tem a seguinte aparência:

      {
          "schema": {
              "dataColumn": [
                  {
                      "name": "mysql_id",
                      "type": "LONG"
                  },
                  {
                      "name": "mysql_title",
                      "type": "STRING"
                  },
                  {
                      "name": "mysql_author",
                      "type": "STRING"
                  },
                  {
                      "name": "submission_date",
                      "type": "DATE"
                  }
              ],
              "primaryKey": [
                  "mysql_id"
              ],
              "source": {
                  "dbType": "MySQL",
                  "dbName": "mysql_to_kafka",
                  "tableName": "mysql_tbl"
              }
          },
          "payload": {
              "before": null,
              "after": {
                  "dataColumn": {
                      "mysql_title": "mysql2kafka",
                      "mysql_author": "tester",
                      "submission_date": 1614700800000
                  }
              },
              "sequenceId": "1614748790461000000",
              "timestamp": {
                  "eventTime": 1614748870000,
                  "systemTime": 1614748870925,
                  "checkpointTime": 1614748870000
              },
              "op": "INSERT",
              "ddl": null
          },
          "version": "0.0.1"
      }

Próximos passos

• Consultar mensagens nos seus tópicos do Kafka

• Habilitar o recurso de conector para instâncias adicionais