ApsaraMQ for Kafka:Criar um Tablestore sink connector

Pré-requisitos

• Você ativou o Tablestore e criou uma instância. Para mais informações, consulte Activate Tablestore and create an instance.

• Você adquiriu e ativou uma instância do ApsaraMQ for Kafka e criou um tópico. Para etapas detalhadas, consulte Purchase an ApsaraMQ for Kafka instance e Create resources.

• A função vinculada ao service gerada pela tarefa do Tablestore sink connector exige a política AliyunOTSFullAccess. Anexe esta política manualmente para conceder à função permissão de gerencie o Tablestore. Para mais informações, consulte Grant permissions to a RAM role.

Etapa 1: Criar uma tabela no Tablestore

Crie uma tabela no Tablestore para armazenar os dados transmitidos do ApsaraMQ for Kafka. Para mais informações, consulte Procedure.

Etapa 2: Criar e iniciar o Tablestore sink connector

1. Faça login no ApsaraMQ for Kafka console. Na página Overview, selecione uma região na seção Resource Distribution.

2. No painel de navegação à esquerda, escolha Connector Ecosystem Integration > Tasks.

3. Na página Tasks, clique em Create Task.

4. Na página Create Task, defina Task Name e Description, configure os parâmetros a seguir e clique em Save.

   • Crie Task

     1. Na etapa Source, selecione Message Queue for Apache Kafka como o Data Provider. Configure os parâmetros a seguir e clique em Next.

        Parameter	Description	Example
        Region	A região da instância de origem Message Queue for Apache Kafka.	China (Beijing)
        Kafka instance	A instância de origem Message Queue for Apache Kafka.	alikafka_post-cn-jte3****
        Topic	O tópico do qual as mensagens são consumidas.	demo-topic
        Group ID	O grupo de consumidores da instância de origem. Quick Create (recomendado): Um Group ID no formato GID_EVENTBRIDGE_xxx é criado automaticamente. Use Existing: Selecione um Group ID existente. Não compartilhe este Group ID com outros services para evitar interrupções no consumo de mensagens.	Quick Create
        Consumer offset	O offset a partir do qual o consumo de mensagens começa. Latest offset (latest) Earliest offset (earliest)	Latest offset (latest)
        Network configuration	O tipo de rede para roteamento de mensagens. Basic Network Self-managed Internet	Basic Network
        VPC	Obrigatório somente se Network configuration estiver definido como Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Obrigatório somente se Network configuration estiver definido como Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        Security group	Obrigatório somente se Network configuration estiver definido como Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	O formato de codificação do conteúdo da mensagem. Recomenda-se Json quando não há requisitos específicos de codificação. Json: Codifica dados binários como um objeto JSON no payload usando UTF-8. Text: Codifica dados binários como uma string UTF-8 no payload. Este é o formato padrão. Binary: Codifica dados binários como uma string Base64 no payload.	Json
        Messages	Parâmetro de Advanced configuration. O número máximo de mensagens por lote. Uma requisição é enviada quando a contagem acumulada de mensagens atinge esse valor. Valores válidos: 1 a 10.000.	100
        Interval (Unit: Seconds)	Parâmetro de Advanced configuration. O intervalo, em segundos, no qual as mensagens são agregadas e enviadas ao sink. Valores válidos: 0 a 15. O valor 0 indica que as mensagens são entregues imediatamente.	3

     2. Na etapa Filtering, defina o Pattern Content para filtrar eventos. Para mais informações, consulte event pattern.

     3. Na etapa Transformation, configure a transformação de dados para operações como divisão, mapeamento, enriquecimento e roteamento dinâmico. Para mais informações, consulte Use Function Compute to clean message data.

     4. Na etapa Sink, defina Service Type como Tablestore e configure os parâmetros a seguir.

        Parameter	Description	Example
        Instance Name	O nome da instância de destino do Tablestore.	ots-sink
        Destination Table	O nome da tabela de dados de destino no Tablestore.	ots_sink_table
        Primary Key	Defina regras de extração para valores de chave primária e colunas de atributo usando a sintaxe JsonPath. Quando o Data Format em Source estiver definido como JSON, os dados de saída do ApsaraMQ for Kafka têm o seguinte formato: { "data": { "topic": "demo-topic", "partition": 0, "offset": 2, "timestamp": 1739756629123, "headers": { "headers": [], "isReadOnly": false }, "key":"ots-sink-k1", "value": "ots-sink-v1" }, "id": "7702ca16-f944-4b08-***-***-0-2", "source": "acs:alikafka", "specversion": "1.0", "type": "alikafka:Topic:Message", "datacontenttype": "application/json; charset=utf-8", "time": "2025-02-17T01:43:49.123Z", "subject": "acs:alikafka:alikafka_serverless-cn-lf6418u6701:topic:demo-topic", "aliyunaccountid": "1******6789" }	Por exemplo, para uma chave primária chamada topic, defina a regra de extração de valor como $.data.topic.
        Attribute Column		Por exemplo, para uma coluna de atributo chamada key, defina a regra de extração de valor como $.data.key.
        Operation Mode	O método de gravação de dados no Tablestore. put: Se já existir um registro com a mesma chave primária, os novos dados substituem os dados existentes. update: Quando dois registros compartilham a mesma chave primária, colunas incrementais são adicionadas à linha sem excluir as colunas existentes. delete: Exclua a linha correspondente à chave primária especificada.	put
        Network configuration	VPC: Entrega mensagens do Kafka ao Tablestore por meio de uma VPC. Public Network: Entrega mensagens do Kafka ao Tablestore pela rede pública.	VPC
        VPC	Selecione o ID da VPC. Este parâmetro é obrigatório somente se Network Configuration estiver definido como VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Selecione o ID do vSwitch. Este parâmetro é obrigatório somente se Network Configuration estiver definido como VPC.	vsw-bp1gbjhj53hdjdkg****
        Security group	Selecione um grupo de segurança. Este parâmetro é obrigatório somente quando Network Configuration estiver definido como VPC.	test_group

   • Task Properties

     Configure a política de novas tentativas para entregas de eventos com falha e o método de tratamento de erros. Para mais informações, consulte Retry policies and dead-letter queues.

5. Volte à página Tasks. Localize sua tarefa e clique em Enable na coluna Actions.

6. Na caixa de diálogo Note, leia as informações e clique em OK.

   Após ativar a tarefa, ela é iniciada em 30 a 60 segundos. Monitore o status na coluna Status da página Tasks.

Etapa 3: Testar o Tablestore sink connector

1. Na página Tasks, na coluna Event Source da sua tarefa, clique no tópico de origem.

2. Na página de detalhes do tópico, clique em Send Test Message.

3. No painel Start to Send and Consume Message, configure a mensagem e clique em OK.

   Na aba Console, defina Message Key como ots-sink-k1, Message Content como ots-sink-v1 e Send to Specified Partition como No.

4. Volte à página Tasks e, na coluna Event Target da sua tarefa, clique no nome da tabela de destino.

5. Na página Manage Table, clique na aba Data Management para visualize os dados na tabela do Tablestore.

   A tabela de dados contém as colunas topic (chave primária), partition (chave primária), offset (chave primária), key e value. Um registro com valores como partition=2, offset=10, key=ots-sink-k1 e value=ots-sink-v1 confirma que a mensagem foi gravada com sucesso no Tablestore.