ApsaraMQ for Kafka:Criar um Tablestore sink connector

Pré-requisitos

• ApsaraMQ for Kafka

  • Ative o recurso de connector para a instância. Para mais informações, consulte Enable connectors.

  • Crie um tópico de source para a instância. Para mais informações, consulte Step 1: Create a topic.

• Tablestore

  • Ative o Tablestore e crie uma instância. Para mais informações, consulte Activate Tablestore and create an instance.

Observações

• Só é possível exportar dados de um tópico de source em uma instância do ApsaraMQ for Kafka para o Tablestore dentro da mesma região. Para informações sobre os limites de connectors, consulte Usage limits.

• Ao criar um connector, o ApsaraMQ for Kafka cria automaticamente uma service-linked role.

  • Se a service-linked role não existir, o ApsaraMQ for Kafka cria uma automaticamente, permitindo que o ApsaraMQ for Kafka exporte dados para o Tablestore.

  • Se a service-linked role já existir, o ApsaraMQ for Kafka não cria uma nova.

  Para mais informações sobre service-linked roles, consulte Service-linked roles.

Procedimento

Este tópico descreve como usar um Tablestore sink connector para exportar dados de um tópico de source em uma instância do ApsaraMQ for Kafka para o Tablestore.

1. Opcional: Crie os tópicos e o group necessários para o Tablestore sink connector.

   Caso não seja necessário personalizar os tópicos e o group, ignore esta etapa e selecione Auto na próxima.

   Importante

   Alguns tópicos exigidos pelo Tablestore sink connector precisam usar o mecanismo de armazenamento Local. Em instâncias do ApsaraMQ for Kafka com versão principal 0.10.2, não é possível criar manualmente tópicos que usem o mecanismo de armazenamento Local. Esses tópicos só podem ser criados automaticamente.

   1. Create the topics for the Tablestore sink connector

   2. Create the group for the Tablestore sink connector

2. Create and deploy the Tablestore sink connector

3. Verifique o resultado

   1. Send a test message

   2. View data in the table

Crie os tópicos para o conector de destino do Tablestore

No console do ApsaraMQ for Kafka, crie manualmente os cinco tópicos necessários para o conector de destino do Tablestore: um tópico de offset de tarefa, um tópico de configuração de tarefa, um tópico de status de tarefa, um tópico de fila de mensagens mortas e um tópico de dados de erro. O número de partições e o mecanismo de armazenamento exigidos variam para cada tópico. Para mais informações, consulte List of source service parameters.

1. Acesse o 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, que corresponde à região onde a instância do ECS está implantada. Tópicos não podem ser compartilhados entre regiões. Por exemplo, se um tópico for criado na região China (Beijing), o produtor e o consumidor de mensagens também precisam estar em execução em uma instância do ECS na região China (Beijing).

3. Na página Instances, clique em no nome da instância de destino.

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	O nome do tópico. Nota No Kafka, nomes de tópico como xxx_xxx e xxx.xxx são considerados idênticos. Um erro será retornado se você tentar criar um tópico com nome duplicado.	demo
   Description	Uma breve descrição do tópico.	demo test
   Partitions	O número de partições do tópico.	12
   Storage Engine Nota No momento, a seleção do tipo de mecanismo de armazenamento está disponível apenas para instâncias que não são da edição Serverless Professional. Para outros tipos de instância, esse parâmetro não é compatível e o valor é definido como Cloud Storage por padrão.	O mecanismo de armazenamento das mensagens do tópico. ApsaraMQ for Kafka é compatível com os dois mecanismos de armazenamento a seguir. Cloud Storage: A camada subjacente utiliza discos da Alibaba Cloud. Esse mecanismo oferece baixa latência, alto desempenho, alta durabilidade e alta confiabilidade, com um mecanismo distribuído de três réplicas. Se a Instance Edition da instância for Standard (High Write), o mecanismo de armazenamento só poderá ser Cloud Storage. Local Storage: Utiliza o algoritmo de replicação ISR (in-sync replica) nativo do Kafka com um mecanismo distribuído de três réplicas.	Cloud Storage
   Message Type	O 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 em que foram enviadas. Se uma máquina do cluster ficar indisponível, as mensagens poderão ser desordenadas. Ao definir Storage Engine como Cloud Storage, a opção Normal Message é selecionada por padrão. Partitionally Ordered 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 uma máquina do cluster fique indisponível, a ordem das mensagens dentro da partição ainda é garantida; no entanto, o envio para algumas partições pode falhar. O funcionamento normal é retomado após a recuperação das partições. Ao definir Storage Engine como Local Storage, a opção Partitionally Ordered Message é selecionada por padrão.	Normal Message
   Log Cleanup Policy	A política de limpeza dos logs do tópico. Ao definir Storage Engine como Local Storage, é obrigatório configurar a Log Cleanup Policy. A opção Local Storage para Storage Engine está disponível apenas para instâncias da edição Professional. Instâncias da edição Standard não são compatíveis. ApsaraMQ for Kafka é compatível com as duas políticas de limpeza a seguir. Delete: A política de limpeza de mensagens padrão. Quando a capacidade do disco é suficiente, as mensagens são retidas pelo período máximo de retenção. Quando a capacidade é insuficiente (geralmente, quando o uso do disco ultrapassa 85%), mensagens antigas são excluídas antecipadamente para garantir a disponibilidade do service. Compact: Aplica a política de limpeza Kafka Log Compaction. Essa política garante que, para mensagens com a mesma chave, o valor mais recente seja sempre preservado. Indicada principalmente para cenários como recuperação de estado após falha do sistema ou recarga de cache após reinicialização. Por exemplo, ao usar Kafka Connect ou Confluent Schema Registry, é necessário utilizar um Compact Topic do Kafka para armazenar o estado do sistema ou informações de configuração. Importante Compact topics geralmente são usados apenas em determinados componentes do ecossistema, como Kafka Connect ou Confluent Schema Registry. Não defina essa propriedade para tópicos em outros cenários de envio e recebimento de mensagens. Para mais informações, consulte a Biblioteca de Demos do ApsaraMQ for Kafka.	Compact
   Tag	As tags do tópico.	demo

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

Crie um group para o Tablestore sink connector

É possível criar manualmente um group para a tarefa de sincronização de dados do Tablestore sink connector no console do ApsaraMQ for Kafka. O nome desse group deve seguir o padrão connect-Nome da Tarefa. Para mais informações, consulte Source service parameters.

1. Acesse o ApsaraMQ for Kafka console.

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

3. Na página Instances, clique em no 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 um nome para o group no campo Group ID, insira uma breve descrição no campo Description, adicione tags ao group e clique em OK.

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

Criar e implantar um Tablestore sink connector

Crie e implante um Tablestore sink connector para sincronizar dados de uma instância do ApsaraMQ for Kafka para uma tabela do Tablestore.

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

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 connector 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 a seguir e clique em Next.

      Parâmetro	Descrição	Exemplo
      Name	O nome do connector. O nome deve atender aos seguintes requisitos: O nome pode ter até 48 caracteres e conter apenas dígitos, letras minúsculas e hifens (-). Não pode começar com um hífen (-). O nome deve ser exclusivo dentro de uma instância do ApsaraMQ for Kafka. A tarefa de sincronização de dados do connector utiliza um consumer group com o nome connect-task-name. Se você não criar esse consumer group manualmente, o sistema o criará automaticamente.	kafka-ts-sink
      Instance	Por padrão, o nome da instância e o ID da instância são exibidos.	demo alikafka_post-cn-st21p8vj****

   2. Na aba Configure Source Service, defina Data Source como ApsaraMQ for Kafka, configure os parâmetros a seguir e clique em Next.

      Nota

      Se você já criou o tópico e o group necessários, selecione Manual para a criação de recursos e insira as informações correspondentes. Caso contrário, selecione Auto.

      Tabela 1. Parâmetros do source service

      Parâmetro	Descrição	Exemplo
      Data Source Topic	O tópico de origem para sincronização de dados.	ts-test-input
      Consumer Thread Concurrency	O número de threads de consumidor simultâneos para o tópico de origem de dados. O valor padrão é 6. Valores válidos: 1 2 3 6 12	6
      Consumer Offset	A posição inicial para o consumo de mensagens. Valores válidos: Earliest Offset: Consome mensagens a partir do offset mais antigo. Latest Offset: Consome mensagens a partir do offset mais recente.	Earliest Offset
      VPC ID	O ID do VPC onde a tarefa de sincronização de dados é executada. Este parâmetro é exibido após clicar em Configure Runtime Environment. Por padrão, utiliza o VPC da instância ApsaraMQ for Kafka, não sendo necessária nenhuma configuração adicional.	vpc-bp1xpdnd3l***
      vSwitch ID	O ID do vSwitch onde a tarefa de sincronização de dados é executada. Este parâmetro é exibido após clicar em Configure Runtime Environment. O vSwitch deve estar no mesmo VPC que a instância ApsaraMQ for Kafka. Por padrão, utiliza o vSwitch especificado durante a implantação da instância ApsaraMQ for Kafka.	vsw-bp1d2jgg81***
      Failure Handling Policy	A política de tratamento de falhas no envio de mensagens em uma partição de tópico. Este parâmetro é exibido após clicar em Configure Runtime Environment. Valores válidos: Continue Subscription: Mantém a assinatura da partição de tópico onde o erro ocorreu e registra logs de erro. Stop Subscription: Interrompe a assinatura da partição de tópico onde o erro ocorreu e registra logs de erro. Nota Para saber como visualizar logs, consulte Manage connectors. Para identificar soluções com base em códigos de erro, consulte Error codes.	Continue Subscription
      Resource Creation Method	O método de criação dos tópicos e group necessários. Este parâmetro é exibido após clicar em Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	O group de consumidor utilizado pela tarefa de sincronização de dados. Este parâmetro é exibido após clicar em Configure Runtime Environment. O nome do group deve seguir o formato connect-task-name.	connect-cluster-kafka-ots-sink
      Task Offset Topic	O tópico que armazena os offsets de consumo. Este parâmetro é exibido após clicar em Configure Runtime Environment. Topic: O nome do tópico deve começar com connect-offset. Partitions: O número de partições do tópico deve ser maior que 1. Storage engine: O mecanismo de armazenamento do tópico deve ser Local Storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-offset-kafka-ots-sink
      Task Configuration Topic	O tópico que armazena as configurações da tarefa. Este parâmetro é exibido após clicar em Configure Runtime Environment. Topic: O nome do tópico deve começar com connect-config. Partitions: O número de partições do tópico deve ser 1. Storage engine: O mecanismo de armazenamento do tópico deve ser Local Storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-config-kafka-ots-sink
      Task Status Topic	O tópico que armazena o status da tarefa. Este parâmetro é exibido após clicar em Configure Runtime Environment. Topic: O nome do tópico deve começar com connect-status. Partitions: O número recomendado de partições é 6. Storage engine: O mecanismo de armazenamento do tópico deve ser Local Storage. cleanup.policy: A política de limpeza de log do tópico deve ser compact.	connect-status-kafka-ots-sink
      Dead-letter Queue Topic	O tópico utilizado para armazenar dados de erro do framework Kafka Connect. Este parâmetro é exibido após clicar em Configure Runtime Environment. Para economizar recursos de tópico, é possível usar o mesmo tópico para a dead-letter queue e o Error Data Topic. Topic: O nome do tópico deve começar com connect-error. Partitions: O número recomendado de partições é 6. Storage engine: O mecanismo de armazenamento do tópico pode ser Local Storage ou Cloud Storage.	connect-error-kafka-ots-sink
      Error Data Topic	O tópico utilizado para armazenar dados de erro do sink. Este parâmetro é exibido após clicar em Configure Runtime Environment. Para economizar recursos de tópico, é possível usar o mesmo tópico para o Dead-letter Queue Topic e o tópico de dados de erro. Topic: O nome do tópico deve começar com connect-error. Partitions: O número recomendado de partições é 6. Storage engine: O mecanismo de armazenamento do tópico pode ser Local Storage ou Cloud Storage.	connect-error-kafka-ots-sink

   3. Na aba Configure Destination Service, defina Destination Service como Tablestore, configure os parâmetros a seguir e, em seguida, clique em Create.

      Parâmetro	Descrição	Exemplo
      Instance Name	O nome da instância do Tablestore.	k00eny67****
      Automatically Create Destination Table	Define se uma tabela deve ser criada automaticamente no Tablestore. Yes: Uma tabela é criada automaticamente no Tablestore para armazenar os dados sincronizados, com base no nome de tabela configurado. No: Os dados sincronizados são armazenados em uma tabela já existente.	Yes
      Destination Table Name	O nome da tabela que armazena os dados sincronizados. Se Automatically Create Destination Table estiver definido como No, o nome da tabela deve ser idêntico ao nome de uma tabela existente na instância do Tablestore.	kafka_table
      Tablestore	O tipo da tabela que armazena os dados sincronizados. Wide Column Model TimeSeries Model	Wide Column Model
      Message Key Format	O formato de entrada da chave da mensagem. Os valores válidos são String e JSON. O valor padrão é JSON. Este parâmetro é exibido somente quando Tablestore está definido como Wide Column Model. String: A chave da mensagem é interpretada diretamente como uma string. JSON: A chave da mensagem deve estar no formato JSON.	String
      Message Value Format	O formato de entrada do valor da mensagem. Os valores válidos são String e JSON. O valor padrão é JSON. Este parâmetro é exibido somente quando Tablestore está definido como Wide Column Model. String: O valor da mensagem é interpretado diretamente como uma string. JSON: O valor da mensagem deve estar no formato JSON.	String
      JSON Message Field Conversion	O método de processamento dos campos em uma mensagem JSON. Este parâmetro é exibido quando Message Key Format ou Message Value Format estiver definido como JSON. Valores válidos: Write All as String: Converte todos os campos para o tipo String no Tablestore. Automatically Identify Field Types: Converte campos do tipo String e Boolean no corpo da mensagem JSON para os tipos String e Boolean correspondentes no Tablestore. Os tipos Integer e Float no corpo da mensagem JSON são convertidos para o tipo Double no Tablestore.	Write All as String
      Primary Key Mode	Define o modo de chave primária. É possível extrair chaves primárias da tabela de diferentes partes dos registros de mensagens do ApsaraMQ for Kafka, incluindo as coordenadas (topic, partition e offset), a chave e o valor. Este parâmetro é exibido somente quando Tablestore está definido como Wide Column Model. O valor padrão é kafka. kafka: Utiliza <connect_topic>_<connect_partition> e <connect_offset> como chaves primárias da tabela de dados. record_key: Utiliza campos da chave do registro como chaves primárias da tabela de dados. record_value: Utiliza campos do valor do registro como chaves primárias da tabela de dados.	kafka
      Primary Key Column Names	Os nomes das colunas de chave primária da tabela de dados e seus tipos de dados correspondentes. Os tipos String e Integer são suportados. Ou seja, os campos da chave ou do valor do registro que corresponderem aos nomes de coluna configurados serão usados como chaves primárias da tabela de dados. Este parâmetro é exibido quando Message Key Format está definido como JSON e Primary Key Mode está definido como record_key, ou quando Message Value Format está definido como JSON e Primary Key Mode está definido como record_value. Clique em Create para adicionar um nome de coluna. É possível configurar no máximo quatro nomes de coluna.	None
      Write Mode	Define o modo de gravação. Os valores válidos são put e update. O valor padrão é put. Este parâmetro é exibido somente quando Tablestore está definido como Wide Column Model. put: Substitui os dados existentes. update: Atualiza os dados existentes.	put
      Delete Mode	Quando um registro de mensagem do ApsaraMQ for Kafka contém valores nulos, este parâmetro determina se linhas ou colunas de atributo devem ser excluídas. Exibido quando Primary Key Mode está definido como record_key. Valores válidos: none: Valor padrão. Nenhuma exclusão é permitida. row: Permite a exclusão de linhas. column: Permite a exclusão de colunas de atributo. row_and_column: Permite a exclusão de linhas e colunas de atributo. O comportamento de exclusão depende do modo de gravação: Quando Write Mode é put, qualquer modo de exclusão resulta em gravações de substituição na tabela de dados do Tablestore, mesmo quando o valor contém campos nulos. Quando Write Mode é update e o modo de exclusão é none ou row, registros em que todos os campos de valor são nulos são tratados como dados inválidos. Se apenas alguns campos de valor forem nulos, o conector ignora os campos nulos e grava os campos não nulos na tabela de dados do Tablestore. Quando o modo de exclusão é column ou row_and_column, o conector exclui linhas e colunas de atributo referentes aos campos nulos e, em seguida, grava os dados na tabela de dados do Tablestore.	None
      Metric Name Field	Este campo é mapeado para o campo de nome de métrica (_m_name) no Tablestore TimeSeries Model. O nome da métrica especifica o nome da grandeza física ou da métrica de monitoramento dos dados de série temporal, como temperatura ou velocidade. Este campo não pode estar vazio. O parâmetro é exibido quando o TimeSeries Model é selecionado para Tablestore.	measurement
      Data Source Field	Mapeie este campo para o campo de fonte de dados (_data_source) no Tablestore TimeSeries Model. Ele atua como identificador da origem dos dados de série temporal, como o nome de uma máquina ou ID de um dispositivo, e pode estar vazio. Este parâmetro é exibido quando o modelo TimeSeries é selecionado para Tablestore.	source
      Tag Field	Use um ou mais campos como campos de tag (_tags) para um Tablestore TimeSeries Model. Cada tag é um par chave-valor de strings. A chave corresponde ao nome do campo configurado e o valor corresponde ao conteúdo do campo. As tags fazem parte dos metadados da série temporal. Uma série temporal é identificada de forma única pela combinação de seu nome de métrica, fonte de dados e tags. As tags podem estar vazias. Este parâmetro é exibido quando Tablestore é selecionado como modelo de série temporal.	tag1, tag2
      Timestamp Field	Mapeia este campo para o campo de timestamp (_time) no Tablestore TimeSeries Model. Ele representa o instante de tempo desta linha de dados de série temporal, como o momento em que uma grandeza física é gerada. Ao gravar dados no Tablestore, o campo de timestamp é convertido para microssegundos. Este parâmetro é exibido quando o TimeSeries Model é selecionado para Tablestore.	time
      Timestamp Unit	Configure este parâmetro de acordo com o campo de timestamp real. Exibido somente quando Tablestore está definido como TimeSeries Model. Valores válidos: SECONDS MILLISECONDS MICROSECONDS NANOSECONDS	MILLISECONDS
      Whether to Map All Non-primary Key Fields	Define se todos os campos que não são chave primária devem ser mapeados como campos de dados. Campos que não são chave primária são aqueles que ainda não foram mapeados como nome de métrica, fonte de dados, tag ou timestamp. Este parâmetro é exibido somente quando Tablestore está definido como TimeSeries Model. Valores válidos: Yes: Os campos são mapeados automaticamente e seus tipos de dados são determinados. Todos os tipos numéricos são convertidos para o tipo Double. No: É necessário especificar os campos e os tipos a serem mapeados.	Yes
      Configure Mapping for All Non-primary Key Fields	Os tipos de dados correspondentes aos nomes dos campos que não são chave primária da tabela de série temporal. Os tipos Double, Integer, String, Binary e Boolean são suportados. Este parâmetro é exibido quando Whether to Map All Non-primary Key Fields está definido como No.	String

      Após a criação do conector, você pode visualizá-lo na página Connectors.

6. Após criar o conector, localize-o na página Connectors e clique em Deploy na coluna Actions.

7. Clique em OK.

Enviar uma mensagem de teste

Após implantar o conector Tablestore sink, envie uma mensagem para o tópico de origem no ApsaraMQ for Kafka para verificar se os dados foram sincronizados com o Tablestore.

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 Sending Method como Console.

     1. No campo Message Key, insira a chave da mensagem. Por exemplo: demo.

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

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

        • Clique em Yes e insira o ID da partição no campo Partition ID. Por exemplo: 0. Para consultar o ID da partição, consulte View partition status.

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

   • Defina 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 Sending Method como SDK. Selecione um SDK para a linguagem ou framework desejado e um tipo de conexão para enviar as mensagens.

Visualizar dados da tabela

Após enviar uma mensagem para o tópico de origem de dados no ApsaraMQ for Kafka, visualize os dados na tabela do Tablestore para confirmar que a mensagem foi recebida.

1. Acesse o console do Tablestore.

2. Na página Overview, clique em o nome da instância ou clique em Instances na coluna Actions.

3. Na guia Instance Details, localize a tabela desejada na seção Tables.

4. Clique no nome da tabela. Na página Table Manage, acesse a guia Data Management para visualizar os dados.