ApsaraMQ for Kafka:Criar um conector sink do AnalyticDB

Pré-requisitos

Antes de começar, verifique se os seguintes pré-requisitos foram atendidos:

• Message Queue for Apache Kafka

  • Ative o recurso de conector na instância do Message Queue for Apache Kafka. Para mais informações, consulte Enable connectors.

  • Crie um tópico de origem para a instância do Message Queue for Apache Kafka. Para mais informações, consulte Step 1: Create a topic.

• Function Compute

  • Activate Function Compute.

• AnalyticDB for MySQL e AnalyticDB for PostgreSQL

  • AnalyticDB for MySQL: No AnalyticDB for MySQL, crie um cluster, crie uma conta de banco de dados, conecte-se ao cluster e, em seguida, crie um banco de dados. Para mais informações, consulte Create a cluster, Create a database account, Connect to a cluster e Create a database.

  • AnalyticDB for PostgreSQL: No console do AnalyticDB for PostgreSQL, crie uma instância, crie uma conta de banco de dados e faça login no banco de dados. Para mais informações, consulte Create an instance, Create and manage accounts e Connect to an instance.

Observações de uso

• A exportação de dados de um tópico de origem de uma instância do Message Queue for Apache Kafka para o AnalyticDB for MySQL ou para o AnalyticDB for PostgreSQL via AnalyticDB for MySQL só é possível se todos os services estiverem na mesma região. Para mais informações sobre os limites do conector, consulte Limits.

• Esse recurso utiliza o Function Compute, que oferece uma cota gratuita. O uso de recursos que exceder essa cota será cobrado. Para mais detalhes, consulte a Billing overview do Function Compute.

• Para solucionar problemas, consulte os logs de chamadas de função no Function Compute. Para mais informações, consulte Configure logging.

• Durante a exportação de dados, o Message Queue for Apache Kafka serializa as mensagens como strings UTF-8. Formatos de dados binários não são suportados.

• Se o conector sink do AnalyticDB utilizar um endpoint privado, o ambiente de execução do Function Compute não poderá acessá-lo por padrão. Para habilitar a conectividade de rede, configure o service da função no console do Function Compute com a mesma VPC e o mesmo vSwitch da sua instância AnalyticDB de destino. Para mais informações, consulte Update a service.

• O Message Queue for Apache Kafka cria automaticamente uma função vinculada ao service quando você cria um conector.

  • Caso não exista uma função vinculada ao service, o Message Queue for Apache Kafka cria uma automaticamente. Essa função concede ao Message Queue for Apache Kafka as permissões necessárias para exportar dados.

  • Se já existir uma função vinculada ao service, o Message Queue for Apache Kafka não criará outra.

  Para mais informações sobre funções vinculadas a services, consulte Service-linked roles.

Procedimento

Para exportar dados de um tópico de origem de uma instância do Message Queue for Apache Kafka para um banco de dados AnalyticDB usando um conector sink do AnalyticDB, siga estas etapas:

1. Opcional: Para o Conector Sink do AnalyticDB, crie o tópico e o grupo necessários

   Se não for necessário personalizar o Tópico e o Grupo, pule esta etapa e opte por criá-los automaticamente na próxima etapa.

   Importante

   Alguns tópicos exigidos pelo conector sink do AnalyticDB devem usar local storage. Em instâncias do Message Queue for Apache Kafka com versão principal 0.10.2, não é possível criar manualmente tópicos que utilizem local storage. Esses tópicos só podem ser criados automaticamente.

   1. Create the topics required by an AnalyticDB sink connector

   2. Create the group required by an AnalyticDB sink connector

2. Create and deploy an AnalyticDB sink connector

3. Configure os services

   1. Configure the Function Compute service

   2. Configure the destination AnalyticDB database

4. Verifique o resultado

   1. Send test messages

   2. Verify the result

Criar os tópicos necessários

É possível criar manualmente os cinco tópicos exigidos pelo conector sink do AnalyticDB no console do Message Queue for Apache Kafka: um tópico de offset de tarefa, um tópico de configuração de tarefa, um tópico de status de tarefa, um dead-letter queue topic e um error data topic. Cada um desses tópicos possui requisitos específicos quanto ao número de partições e ao mecanismo de armazenamento. Para mais detalhes, consulte a tabela Parameters for configuring the source service.

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

2. Na página Overview, selecione uma região na seção Resource Distribution.

   Importante

   Os tópicos devem ser criados na mesma região da sua aplicação, ou seja, onde a instância ECS está implantada. Não é possível utilizar tópicos entre regiões diferentes. Por exemplo, se um tópico for criado na região China (Beijing), tanto o produtor quanto o consumidor de mensagens deverão rodar em uma instância ECS também localizada na região China (Beijing).

3. Na página Instances, clique em nome da instância desejada.

4. No painel de navegação à esquerda, clique em Topics.

5. Na página Topics, clique em Create Topic.

6. No painel Create Topic, defina as propriedades do tópico e clique em OK.

   Parâmetro	Descrição	Exemplo
   Name	Nome do tópico. Nota No Kafka, nomes de tópicos como xxx_xxx e xxx.xxx são considerados idênticos. Um erro será reportado caso você tente criar um tópico com um nome duplicado.	demo
   Description	Breve descrição do tópico.	demo test
   Partitions	Quantidade de partições no tópico.	12
   Storage Engine Nota Atualmente, a seleção do tipo de mecanismo de armazenamento está disponível apenas para instâncias da Professional Edition não Serverless. Para outras instâncias, este parâmetro não é suportado e assume o valor padrão Cloud Storage.	Mecanismo de armazenamento das mensagens do tópico. ApsaraMQ for Kafka suporta os dois mecanismos de armazenamento a seguir. Cloud Storage: A camada subjacente utiliza discos da Alibaba Cloud. Este mecanismo oferece baixa latência, alto desempenho, alta durabilidade e alta confiabilidade, empregando um mecanismo distribuído de três réplicas. Se a Instance Edition da instância for Standard (High Write), o único mecanismo de armazenamento disponível será o Cloud Storage. Local Storage: Utiliza o algoritmo nativo de replicação ISR (in-sync replica) do Kafka e um mecanismo distribuído de três réplicas.	Cloud Storage
   Message Type	Tipo das mensagens do tópico. Normal Message: Por padrão, mensagens com a mesma chave são distribuídas para a mesma partição, e as mensagens dentro de uma partição são armazenadas na ordem de envio. Caso ocorra falha em uma máquina do cluster, a ordem das mensagens pode ser perdida. Ao definir o Storage Engine como Cloud Storage, a opção Normal Message é selecionada automaticamente. Partitionally Ordered Message: Por padrão, mensagens com a mesma chave vão para a mesma partição, mantendo a ordem de envio dentro dela. Mesmo se houver falha em uma máquina do cluster, a ordenação dentro da partição é garantida. No entanto, o envio para algumas partições pode falhar temporariamente, retomando a normalidade após a recuperação. Quando o Storage Engine é definido como Local Storage, a opção Partitionally Ordered Message vem selecionada por padrão.	Normal Message
   Log Cleanup Policy	Política de limpeza dos logs do tópico. Ao configurar o Storage Engine como Local Storage, é obrigatório definir a Log Cleanup Policy. A definição do Storage Engine como Local Storage é permitida apenas em instâncias da Professional Edition; instâncias da Standard Edition não suportam essa configuração. ApsaraMQ for Kafka oferece as duas políticas de limpeza abaixo. Delete: Política padrão de limpeza de mensagens. Se houver espaço suficiente em disco, as mensagens são retidas pelo período máximo configurado. Caso o espaço seja insuficiente (geralmente quando o uso do disco ultrapassa 85%), mensagens antigas são excluídas antecipadamente para preservar a disponibilidade do service. Compact: Implementa a política de Log Compaction do Kafka. Essa política garante que, para mensagens com a mesma chave, apenas o valor mais recente seja mantido. É ideal para cenários como recuperação de estado após falhas ou recarga de cache após reinicialização do sistema. Por exemplo, ao usar Kafka Connect ou Confluent Schema Registry, é necessário utilizar um Kafka Compact Topic para armazenar estados do sistema ou configurações. Importante Tópicos compactados destinam-se geralmente a componentes específicos do ecossistema, como Kafka Connect ou Confluent Schema Registry. Evite aplicar essa propriedade em tópicos usados para outros cenários de envio e recebimento de mensagens. Para mais detalhes, consulte a Biblioteca de Demos do ApsaraMQ for Kafka.	Compact
   Tag	Tags associadas ao tópico.	demo

   Após a criação, o tópico aparecerá na lista da página Topics.

Criar o grupo necessário para o Conector Sink do AnalyticDB

Você pode criar manualmente o grupo para a tarefa de sincronização de dados do Conector Sink do AnalyticDB no console do Message Queue for Apache Kafka. O nome do grupo deve seguir o formato connect-Nome da Tarefa. Para mais informações, consulte Source service parameters.

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

2. Na página Overview, selecione uma região na seção Resource Distribution.

3. Na página Instances, clique em nome da instância desejada.

4. No painel de navegação à esquerda, clique em Groups.

5. Na página Groups, clique em Create Group.

6. No painel Create Group, insira o nome do grupo na caixa de texto Group ID, adicione uma breve descrição na caixa Description, atribua tags ao grupo e clique em OK.

   Após a criação, o grupo será exibido na lista da página Groups.

Criar e implantar um conector sink do AnalyticDB

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

2. Na página Overview, selecione uma região na seção Resource Distribution.

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

4. Na página Connectors, selecione a instância à qual o conector pertence na lista suspensa Select Instance e clique em Create Connector.

5. No assistente Create Connector, conclua as etapas a seguir.

   1. Na aba Configure Basic Information, configure os parâmetros conforme necessário e clique em Next.

      Parâmetro	Descrição	Valor de exemplo
      Name	Nome do conector. Regras de nomenclatura: O nome deve ter entre 1 e 48 caracteres. O primeiro caractere pode ser uma letra maiúscula, minúscula ou dígito. Os demais podem ser letras minúsculas, dígitos ou hífens (-). O nome precisa ser único dentro da mesma instância do Message Queue for Apache Kafka. Uma tarefa de sincronização de dados de um Connector requer um Grupo chamado connect-nome da tarefa. Se você não criar o Grupo manualmente, o sistema o criará automaticamente.	kafka-adb-sink
      Instance	O nome e o ID da instância são preenchidos automaticamente.	demo alikafka_post-cn-st21p8vj****

   2. Na aba Configure Source Service, selecione Message Queue for Apache Kafka como Data Source, configure os parâmetros abaixo e clique em Next.

      Tabela 1. Parâmetros do service de origem

      Parâmetro	Descrição	Exemplo
      Data Source Topic	Tópico de onde os dados serão sincronizados.	adb-test-input
      Consumer Thread Concurrency	Número de threads simultâneas de consumo para o tópico de origem. Valor padrão: 6. Valores válidos: 1 2 3 6 12	6
      Consumer Offset	Posição inicial do consumo. Valores válidos: Earliest Offset: Inicia o consumo a partir do offset mais antigo disponível. Latest Offset: Inicia o consumo a partir do offset mais recente disponível.	Earliest Offset
      VPC ID	VPC onde a tarefa de sincronização será executada. Este parâmetro aparece após clicar em Configure Runtime Environment. Por padrão, corresponde à VPC da instância do Message Queue for Apache Kafka, não exigindo entrada manual.	vpc-bp1xpdnd3l***
      vSwitch ID	vSwitch onde a tarefa de sincronização será executada. Este parâmetro aparece após clicar em Configure Runtime Environment. O vSwitch deve pertencer à mesma VPC da instância do Message Queue for Apache Kafka. Por padrão, corresponde ao vSwitch especificado durante a implantação da instância do Message Queue for Apache Kafka.	vsw-bp1d2jgg81***
      Failure Handling Policy	Política para lidar com partições onde ocorre falha no envio de mensagens. Este parâmetro aparece após clicar em Configure Runtime Environment. Valores válidos: Continue Subscription: Mantém a assinatura da partição e registra um log de erro. Stop Subscription: Interrompe a assinatura da partição e registra um log de erro. Nota Para saber como visualizar logs, consulte Connector-related operations. Para encontrar soluções baseadas em códigos de erro, consulte Error codes.	Continue Subscription
      Resource Creation Method	Método de criação dos tópicos e do grupo necessários para o conector. Este parâmetro aparece após clicar em Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	Grupo utilizado pela tarefa de sincronização de dados do conector. Clique em Configure Runtime Environment para exibir este parâmetro. O nome do grupo deve ser connect-nome da tarefa.	connect-kafka-adb-sink
      Task Offset Topic	Tópico usado para armazenar offsets de consumo. Este parâmetro aparece após clicar em Configure Runtime Environment. Tópico: Recomenda-se que o nome comece com connect-offset. Partições: A quantidade deve ser maior que 1. Storage Engine: O tópico deve usar local storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-offset-kafka-adb-sink
      Task Configuration Topic	Tópico usado para armazenar configurações da tarefa. Este parâmetro aparece após clicar em Configure Runtime Environment. Tópico: Recomenda-se que o nome comece com connect-config. Partições: A quantidade deve ser exatamente 1. Storage Engine: O tópico deve usar local storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-config-kafka-adb-sink
      Task Status Topic	Tópico usado para armazenar o status da tarefa. Este parâmetro aparece após clicar em Configure Runtime Environment. Tópico: Recomenda-se que o nome comece com connect-status. Partições: Recomenda-se utilizar 6 partições. Storage Engine: O tópico deve usar local storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-status-kafka-adb-sink
      Dead-letter Queue Topic	Tópico destinado a armazenar dados de erro do framework Connect. Clique em Configure Runtime Environment para exibir este parâmetro. Para economizar recursos, este tópico pode ser o mesmo que o error data topic. Tópico: Recomenda-se que o nome comece com connect-error. Partições: Recomenda-se utilizar 6 partições. Storage Engine: Pode utilizar local storage ou cloud storage.	connect-error-kafka-adb-sink
      Error Data Topic	Tópico destinado a armazenar dados de exceção do Sink. Clique em Configure Runtime Environment para exibir este parâmetro. Para economizar recursos, este tópico pode ser o mesmo que o dead-letter queue topic. Tópico: Recomenda-se que o nome comece com connect-error. Partições: Recomenda-se utilizar 6 partições. Storage Engine: Pode utilizar local storage ou cloud storage.	connect-error-kafka-adb-sink

   3. Na aba Configure Destination Service, selecione AnalyticDB como Destination Service, configure os parâmetros abaixo e clique em Create.

      Parâmetro	Descrição	Exemplo
      Instance Type	Tipo da instância AnalyticDB de destino. Valores válidos: AnalyticDB for MySQL e AnalyticDB for PostgreSQL.	MySQL
      AnalyticDB Instance ID	ID da instância de destino do AnalyticDB for MySQL ou do AnalyticDB for PostgreSQL.	am-bp139yqk8u1ik****
      Database Name	Nome do banco de dados de destino na instância AnalyticDB.	adb_demo
      Table Name	Nome da tabela de destino no banco de dados AnalyticDB onde as mensagens serão armazenadas.	user
      Database Username	Nome de usuário para conectar à instância AnalyticDB e importar dados.	adbmysql
      Database Password	Senha para conectar à instância AnalyticDB e importar dados. A senha é definida durante a criação da instância. Caso tenha esquecido, é possível redefini-la. AnalyticDB for MySQL: Para instruções sobre como redefinir a senha, consulte Change an account password. AnalyticDB for PostgreSQL: Acesse o console do AnalyticDB for PostgreSQL, clique em nome da instância e, no painel de navegação à esquerda, clique em Account Management. Localize a conta cuja senha deseja redefinir e clique em Reset Password na coluna Actions.	****

      Nota

      O Message Queue for Apache Kafka transmite o nome de usuário e a senha como variáveis de ambiente para a função do Function Compute durante a criação da tarefa. Após a criação, o Message Queue for Apache Kafka não armazena essas informações.

      Após a criação, o conector estará visível na página Connectors.

6. Na página Connectors, localize o conector criado e clique em Deploy na coluna Actions.

Configurar o service do Function Compute

Após criar e implantar com sucesso um Conector Sink do AnalyticDB no console do Message Queue for Apache Kafka, o Function Compute gera automaticamente um service e uma função para o conector. O service recebe um nome no formato kafka-service-<nome_do_conector>-<string_aleatória>, enquanto a função segue o padrão fc-adb-<string_aleatória>.

1. Na página Connectors, localize o conector desejado e clique em Configure Function na coluna Actions.

   Essa ação redireciona você para o console do Function Compute.

2. No console do Function Compute, localize o service de função criado automaticamente e configure suas informações de VPC e vSwitch. Para etapas detalhadas, consulte Update a service.

Configurar o banco de dados AnalyticDB de destino

Após configurar o service do Function Compute, adicione o bloco CIDR do seu vSwitch à lista de permissões da sua instância AnalyticDB. Esse bloco CIDR pode ser encontrado na página vSwitch do console da VPC, na linha correspondente à VPC e ao vSwitch do service do Function Compute.

• AnalyticDB for MySQL: Acesse o console do AnalyticDB for MySQL e configure a lista de permissões. Para mais informações, consulte Configure an IP address whitelist.

• AnalyticDB for PostgreSQL: Acesse o console do AnalyticDB for PostgreSQL e configure a lista de permissões. Para mais informações, consulte Configure a whitelist.

Enviar mensagens de teste

1. Na página Connectors, localize o conector desejado e clique em Test na coluna Actions.

2. No painel Send Message, envie uma mensagem de teste.

   • Defina o Sending Method como Console.

     1. Na caixa de texto Message Key, insira a chave da mensagem. Por exemplo, demo.

     2. Na caixa de texto Message Content, insira o conteúdo da mensagem de teste. Por exemplo, {"key": "test"}.

     3. Configure a opção Send to Specified Partition para definir se a mensagem será enviada a uma partição específica.

        • Clique em Yes e insira o ID da partição na caixa de texto Partition ID. Por exemplo, 0. Para consultar o ID da partição, veja View partition status.

        • Clique em No para não especificar uma partição.

   • Defina o Sending Method como Docker. Execute o comando Docker na seção Run the Docker container to produce a sample message para enviar a mensagem.

   • Defina o Sending Method como SDK. Selecione um SDK para a linguagem ou framework desejado e um tipo de conexão para enviar mensagens.

Verificar o resultado

Após enviar mensagens para o tópico de origem no Message Queue for Apache Kafka, faça login no console do AnalyticDB for MySQL ou no console do AnalyticDB for PostgreSQL, conecte-se ao banco de dados e acesse a SQL Window do Data Management Service (DMS). Consulte a tabela na instância correspondente para confirmar se os dados foram exportados corretamente.