Criar um conector sink do MaxCompute - ApsaraMQ for Kafka: Central de Documentação do Alibaba Cloud

Criar uma função RAM

As funções do Resource Access Management (RAM) não suportam a seleção direta do ApsaraMQ for Kafka como service confiável. Portanto, ao criar uma função RAM, selecione qualquer service suportado como confiável. Depois que a função RAM for criada, modifique manualmente a política de confiança.

Faça login no console RAM.
Na barra de navegação à esquerda, escolha Identities > Roles.
Na página Roles, clique em Create Role.

A figura a seguir mostra o botão Create Role na página Roles.
No painel Create Role, execute as operações a seguir.
Defina o tipo de entidade confiável como Alibaba Cloud Service e clique em Next.
Na seção Role Type, selecione Normal Service Role. No campo Role Name, insira AliyunKafkaMaxComputeUser1. Na lista Select Trusted Service, selecione MaxCompute e clique em Finish.
Na página Roles, localize AliyunKafkaMaxComputeUser1 e clique em AliyunKafkaMaxComputeUser1.
Na página AliyunKafkaMaxComputeUser1, clique na aba Trust Policy e depois em Edit Trust Policy.
No painel Edit Trust Policy, substitua odps no script por alikafka e clique em OK.

A política a seguir mostra o resultado após a substituição.

Trust policy of the RAM role after odps is replaced with alikafka

Adicionar permissões

Para permitir que o conector sincronize mensagens com uma tabela do MaxCompute, conceda pelo menos as seguintes permissões à função RAM criada:

Objeto	Ação	Description
Projeto	CreateInstance	Cria uma instância no projeto.
Tabela	Describe	Lê os metadados da tabela.
Tabela	Alter	Modifica os metadados da tabela ou adiciona e exclui partições.
Tabela	Update	Sobrescreve ou adiciona dados à tabela.

Para mais informações sobre as permissões anteriores e como concedê-las, consulte Permissões do MaxCompute.

Os passos a seguir mostram como conceder permissões à função AliyunKafkaMaxComputeUser1 criada neste tópico. O MaxCompute gerencia uma função RAM como um usuário, portanto, cada comando nesta seção tem como alvo o objeto RAM$<accountid>:role/aliyunkafkamaxcomputeuser1.

Nota

Em cada um dos comandos a seguir, substitua pelo seu próprio ID de conta Alibaba Cloud.

Faça login no cliente do MaxCompute.
Execute o comando a seguir para adicionar a função RAM como usuário.
```
add user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;
```
Conceda à função RAM o privilégio mínimo necessário para acessar o MaxCompute.
Execute o comando a seguir para conceder à função RAM as permissões no projeto.

grant CreateInstance on project connector_test to user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

Execute o comando a seguir para conceder à função RAM as permissões na tabela.

grant Describe, Alter, Update on table test_kafka to user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

Criar os tópicos necessários para um conector sink do MaxCompute

No console do ApsaraMQ for Kafka, crie manualmente os cinco tópicos exigidos por um conector sink do MaxCompute: tópico de offset da tarefa, tópico de configuração da tarefa, tópico de status da tarefa, tópico de fila de mensagens mortas e tópico de dados de erro. O número necessário de partições e o mecanismo de armazenamento variam conforme o tópico, conforme descrito na tabela a seguir.

Tópico	Prefixo de nome recomendado	Partições	Mecanismo de armazenamento	cleanup.policy
Tópico de offset da tarefa	connect-offset	Maior que 1	Armazenamento local	compact
Tópico de configuração da tarefa	connect-config	1	Armazenamento local	compact
Tópico de status da tarefa	connect-status	6 recomendado	Armazenamento local	compact
Tópico de fila de mensagens mortas	connect-error	6 recomendado	Armazenamento local ou cloud storage	—
Tópico de dados de erro	connect-error	6 recomendado	Armazenamento local ou cloud storage	—

Para economizar recursos de tópicos, utilize um único tópico tanto para a fila de mensagens mortas quanto para os dados de erro. Para esses dois casos, o mecanismo de armazenamento pode ser local ou cloud storage, e as propriedades restantes seguem as regras gerais descritas na tabela de propriedades de tópicos desta seção. Para a descrição completa de cada tópico, consulte {{XREF_15}}.

Faça login no console do ApsaraMQ for Kafka.

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

Importante

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

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

Importante

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

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

Importante

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

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

Importante

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

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

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

Na página Topics, clique em Create Topic.

No painel

Create Topic

, defina as propriedades do tópico e clique em

OK

.

Configure as propriedades de cada tópico do conector com os valores listados na tabela anterior. A tabela a seguir descreve todas as propriedades do tópico.

Parâmetro	Descrição	Exemplo
Name	Nome do tópico. No Kafka, nomes de tópicos nos formatos `xxx_xxx` e `xxx.xxx` são considerados o mesmo nome. Se você criar um tópico duplicado, uma mensagem de erro aparecerá.	connect-offset-kafka-maxcompute-sink
Description	Breve descrição do tópico.	demo test
Partitions	Quantidade de partições do tópico.	12
Storage Engine > Nota: Apenas instâncias Professional Edition não serverless permitem selecionar o tipo de mecanismo de armazenamento. Outras instâncias não oferecem essa opção e usam o tipo cloud storage por padrão.	Mecanismo de armazenamento das mensagens do tópico. O ApsaraMQ for Kafka suporta os dois mecanismos a seguir. - Cloud Storage: Utiliza discos da Alibaba Cloud na camada subjacente, oferecendo baixa latência, alto desempenho, durabilidade e alta confiabilidade. Este mecanismo usa um sistema distribuído de três réplicas. Se a Instance Edition da instância for Standard (High Write), o mecanismo de armazenamento poderá ser apenas Cloud Storage. - Local Storage: Usa o algoritmo de replicação In-Sync Replicas (ISR) do Kafka open source e um mecanismo distribuído de três réplicas.	Local 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. Se houver falha em uma máquina do cluster, as mensagens podem ficar fora de ordem. Se você definir o Storage Engine como Cloud Storage, a opção Normal Message será selecionada por padrão. - Partitionally Ordered Message: Por padrão, mensagens com a mesma chave vão para a mesma partição e são armazenadas na ordem de envio. Mesmo se uma máquina do cluster falhar, as mensagens na partição continuam sendo armazenadas na ordem original. Contudo, o envio para algumas partições pode falhar até que elas se recuperem. Se você definir o Storage Engine como Local Storage, a opção Partitionally Ordered Message será selecionada por padrão.	Normal Message
Log Cleanup Policy	Política de limpeza de logs do tópico. Caso defina o Storage Engine como Local Storage, configure obrigatoriamente a Log Cleanup Policy. O ApsaraMQ for Kafka oferece duas políticas de limpeza: - Delete: política padrão. Se houver espaço suficiente em disco, as mensagens ficam retidas durante o período máximo. Quando o espaço for insuficiente (geralmente acima de 85% de uso), mensagens mais antigas são excluídas antecipadamente para garantir a disponibilidade do service. - Compact: adota a política de compactação de log do Kafka. Essa política garante a retenção do valor mais recente de cada chave de mensagem. É ideal para cenários como restauração de estado após falhas ou recarga de cache após reinicialização do sistema. Por exemplo, ao usar Confluent Schema Registry ou Kafka Connect, é necessário usar um tópico compactado do Kafka para armazenar o estado do sistema ou informações de configuração. > Importante: Tópicos compactados destinam-se geralmente a componentes específicos do ecossistema, como Confluent Schema Registry ou Kafka Connect. Não ative essa propriedade em tópicos usados para enviar e receber mensagens em outros cenários. Para mais informações, consulte a biblioteca de demonstrações do ApsaraMQ for Kafka.	Compact
Tag	Tag associada ao tópico.	demo

Após a criação, o tópico aparecerá na lista da página Topics. Como estes tópicos foram criados manualmente, defina o Resource Creation Method como Manual ao criar o conector e insira os nomes dos tópicos criados.

Criar o Group necessário para um conector sink do MaxCompute

No console do ApsaraMQ for Kafka, crie manualmente o Group utilizado pela tarefa de sincronização de dados de um conector sink do MaxCompute. O nome do Group deve ser connect-nome da tarefa, onde nome da tarefa corresponde ao nome do conector. Para mais informações, consulte {{XREF_16}}.

Faça login no console do ApsaraMQ for Kafka.

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

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

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

Na página Groups, clique em Create Group.

No painel

Create Group

, insira o nome do

Group

no campo

Group ID

, adicione uma breve descrição no campo

Description

, inclua tags e clique em

OK

.

Depois de criado, o Group aparecerá na lista da página Groups. Insira esse nome no campo Connector Consumer Group durante a criação do conector.

Criar e implantar um conector sink do MaxCompute

Crie e implante um conector sink do MaxCompute para sincronizar dados do ApsaraMQ for Kafka com o MaxCompute.

Faça login no console do ApsaraMQ for Kafka.

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

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

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

Na página do assistente de configuração Create Connector, realize as operações a seguir.

Na aba Configure Basic Information, configure os parâmetros abaixo conforme suas necessidades e clique em Next.

Parâmetro	Descrição	Exemplo
Name	Nome do conector. Regras de nomenclatura: - Pode conter dígitos, letras minúsculas e hífens (-), mas não pode começar com hífen. Deve ter entre 1 e 48 caracteres. - Precisa ser único dentro da mesma instância do ApsaraMQ for Kafka. A tarefa de sincronização de dados do conector deve utilizar um Group chamado connect-nome da tarefa. Caso não crie o Group manualmente, o sistema o criará automaticamente.	kafka-maxcompute-sink
Instance	Por padrão, o nome e o ID da instância já vêm especificados.	demo alikafka_post-cn-st21p8vj****

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

Nota

Se já tiver criado um tópico e um consumer group, opte pela criação manual de recursos e informe os dados dos seus recursos existentes. Caso contrário, escolha a criação automática.

Parameter	Description	Example
Data Source Topic	Tópico source cujos dados serão sincronizados.	maxcompute-test-input
Consumer Thread Concurrency	Número de threads simultâneas de consumo para o tópico source. Valor padrão: 6. Valores válidos: - 1 - 2 - 3 - 6 - 12	6
Consumer Offset	Offset inicial para consumo das mensagens. Valores válidos: - Earliest Offset: O consumo começa pelo offset mais antigo. - Latest Offset: O consumo começa pelo offset mais recente.	Earliest Offset
VPC ID	VPC onde a tarefa de sincronização de dados será executada. Clique em Configure Runtime Environment para exibir este parâmetro. Por padrão, utiliza-se a VPC onde a instância ApsaraMQ for Kafka está implantada. Não é necessário especificar este parâmetro.	vpc-bp1xpdnd3l***
vSwitch ID	vSwitch onde a tarefa de sincronização de dados será executada. Clique em Configure Runtime Environment para exibir este parâmetro. O vSwitch deve estar na mesma VPC da instância ApsaraMQ for Kafka. Por padrão, usa-se o vSwitch especificado durante a implantação da instância ApsaraMQ for Kafka.	vsw-bp1d2jgg81***
Failure Handling Policy	Define se o sistema continua assinando a partição do tópico onde ocorreu um erro após a falha no envio de uma mensagem. Clique em Configure Runtime Environment para exibir este parâmetro. Valores válidos: - Continue Subscription: Continua a assinatura da partição com erro e registra o log de erro. - Stop Subscription: Interrompe a assinatura da partição com erro e registra o log de erro > Nota: - Para saber como visualizar logs, consulte Operações do conector. - Para encontrar soluções com base em códigos de erro, consulte Códigos de erro. - Para retomar a assinatura da partição com erro, para entrar em contato com o suporte técnico do .	Continue Subscription
Resource Creation Method	Método usado para criar os tópicos e o Group exigidos pelo conector. Clique em Configure Runtime Environment para exibir este parâmetro. Valores válidos: - Auto: O sistema cria automaticamente os tópicos e o Group necessários. Escolha esta opção se não os tiver criado previamente. - Manual: Selecione esta opção caso tenha criado os tópicos e o Group antecipadamente e insira seus nomes nos campos correspondentes.	Auto
Connector Consumer Group	Group utilizado pela tarefa de sincronização de dados do conector. Clique em Configure Runtime Environment para exibir este parâmetro. O nome do Group deve seguir o padrão connect-nome da tarefa.	connect-kafka-maxcompute-sink
Task Offset Topic	Tópico que armazena os offsets de consumo. Clique em Configure Runtime Environment para exibir este parâmetro. - Nome do tópico: Recomenda-se iniciar com connect-offset. - Número de partições: Deve ser maior que 1. - Mecanismo de armazenamento: Obrigatoriamente armazenamento local. - cleanup.policy: A política de limpeza de logs deve ser compact.	connect-offset-kafka-maxcompute-sink
Task Configuration Topic	Tópico destinado às configurações da tarefa. Clique em Configure Runtime Environment para exibir este parâmetro. - Nome do tópico: Recomenda-se iniciar com connect-config. - Número de partições: Deve ser exatamente 1. - Mecanismo de armazenamento: Obrigatoriamente armazenamento local. - cleanup.policy: A política de limpeza de logs deve ser compact.	connect-config-kafka-maxcompute-sink
Task Status Topic	Tópico que guarda o status da tarefa. Clique em Configure Runtime Environment para exibir este parâmetro. - Nome do tópico: Recomenda-se iniciar com connect-status. - Número de partições: Sugere-se definir como 6. - Mecanismo de armazenamento: Obrigatoriamente armazenamento local. - cleanup.policy: A política de limpeza de logs deve ser compact.	connect-status-kafka-maxcompute-sink
Dead-letter Queue Topic	Tópico para armazenar dados anômalos do framework Connect. Clique em Configure Runtime Environment para exibir este parâmetro. Para poupar recursos, este tópico pode servir também como tópico de dados de erro. - Nome do tópico: Recomenda-se iniciar com connect-error. - Número de partições: Sugere-se definir como 6. - Mecanismo de armazenamento: Pode ser armazenamento local ou cloud storage.	connect-error-kafka-maxcompute-sink
Error Data Topic	Tópico para armazenar dados anômalos do sink. Clique em Configure Runtime Environment para exibir este parâmetro. Para poupar recursos, este tópico pode servir também como fila de mensagens mortas. - Nome do tópico: Recomenda-se iniciar com connect-error. - Número de partições: Sugere-se definir como 6. - Mecanismo de armazenamento: Pode ser armazenamento local ou cloud storage.	connect-error-kafka-maxcompute-sink

Na aba Configure Destination Service, defina Destination Service como MaxCompute, configure os parâmetros abaixo e clique em Create.

Nota

Se a instância do conector estiver nas regiões China (Hangzhou) ou China (Chengdu), uma caixa de diálogo Service Authorization surgirá para cada uma das funções vinculadas ao service AliyunServiceRoleForEventBridgeSourceKafka e AliyunServiceRoleForEventBridgeConnectVPC quando você definir Destination Service como MaxCompute. Na caixa de diálogo Service Authorization, clique em OK. Em seguida, configure os parâmetros a seguir e clique em Create. Caso as funções vinculadas ao service já existam, elas não serão recriadas e a caixa de diálogo Service Authorization não aparecerá.

Parâmetro	Descrição	Exemplo
Endpoint	Endpoint do service MaxCompute. Substitua o ID da região no endpoint pelo ID da região do seu projeto MaxCompute. Para mais informações, consulte endpoints. - Endpoint de VPC: Baixa latência e recomendado. Use quando a instância do ApsaraMQ for Kafka e o MaxCompute estiverem na mesma região. - Endpoint público: Alta latência e não recomendado. Use apenas se a instância do ApsaraMQ for Kafka e o MaxCompute estiverem em regiões diferentes. Para utilizar um endpoint público, ative o acesso à rede pública para o conector. Para mais informações, consulte Ativar acesso à rede pública para um conector.	http://service.cn-hangzhou.maxcompute.aliyun-inc.com/api
Workspace	Workspace do MaxCompute, correspondente ao projeto que contém a tabela de destino.	connector_test
Table	Tabela de destino no MaxCompute.	test_kafka
Region for Table	Região onde a tabela do MaxCompute está localizada.	China (Hangzhou)
Alibaba Cloud Account ID	ID da conta Alibaba Cloud associada ao MaxCompute.	188***
RAM Role	Nome da função RAM do ApsaraMQ for Kafka. Para mais informações, consulte {{XREF_17}}.	AliyunKafkaMaxComputeUser1
Mode	Modo de sincronização das mensagens com o conector. Valor padrão: DEFAULT. Valores válidos: - KEY: Retém apenas a chave da mensagem e a grava na coluna key da tabela do MaxCompute. - VALUE: Retém apenas o valor da mensagem e o grava na coluna value da tabela do MaxCompute. - DEFAULT: Retém tanto a chave quanto o valor da mensagem, gravando-os nas colunas key e value da tabela do MaxCompute. > Importante: No modo DEFAULT, o formato CSV não é suportado. Apenas os formatos TEXT e BINARY são aceitos.	DEFAULT
Format	Formato de sincronização das mensagens com o conector. Valor padrão: TEXT. Valores válidos: - TEXT: As mensagens são strings. - BINARY: As mensagens são arrays de bytes. - CSV: As mensagens são strings separadas por vírgulas (,). > Importante: No formato CSV, o modo DEFAULT não é suportado. Apenas os modos KEY e VALUE são aceitos: - Modo KEY: Retém apenas a chave da mensagem. A string da chave é separada por vírgulas (,) e as partes resultantes são gravadas na tabela seguindo a ordem de seus índices. - Modo VALUE: Retém apenas o valor da mensagem. A string do valor é separada por vírgulas (,) e as partes resultantes são gravadas na tabela seguindo a ordem de seus índices.	TEXT
Partition	Granularidade da partição. Valor padrão: HOUR. Valores válidos: - DAY: Os dados são gravados em uma nova partição diariamente. - HOUR: Os dados são gravados em uma nova partição a cada hora. - MINUTE: Os dados são gravados em uma nova partição a cada minuto.	HOUR
Time Zone	Fuso horário do cliente produtor do ApsaraMQ for Kafka que envia mensagens para o tópico source do conector. Valor padrão: GMT+08:00.	GMT+08:00

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

Depois de criar o conector, acesse a página Connectors, localize o conector criado e clique em Deploy na coluna Actions.

Enviar uma mensagem de teste

Após implantar o conector sink do MaxCompute, envie uma mensagem para o tópico source do ApsaraMQ for Kafka para validar se a sincronização de dados com o MaxCompute está funcionando.

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

No painel Send Message, envie uma mensagem de teste.

Defina o Sending Method como Console.

Na caixa de texto Message Key, insira a chave da mensagem, por exemplo, demo.
No campo Message Content, insira o conteúdo da mensagem de teste, por exemplo, {"key": "test"}.
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 no campo Partition ID, por exemplo, 0. Para saber como consultar o ID da partição, consulte Visualizar status da partição.
Clique em No para enviar a mensagem sem especificar uma partição.

Defina o Sending Method como Docker e 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. Conforme suas necessidades, selecione o kit de desenvolvimento de software (SDK) da linguagem de programação ou framework desejado e o método de conexão, e então envie mensagens usando o SDK.

Visualizar dados da tabela

Após enviar mensagens para o tópico source do ApsaraMQ for Kafka, visualize os dados da tabela no cliente do MaxCompute para confirmar o recebimento das mensagens.

Os passos a seguir mostram como visualizar os dados gravados em test_kafka neste tópico. O conector grava cada mensagem na partição correspondente ao momento da escrita, respeitando a granularidade definida no parâmetro Partition.

Faça login no cliente do MaxCompute.
Execute o comando a seguir para visualizar as partições de dados da tabela.
```
show partitions test_kafka;
```
O seguinte resultado é retornado:
```
pt=11-17-2020 15

OK
```
Execute o comando a seguir para visualizar os dados na partição retornada na etapa anterior.

select * from test_kafka where pt ="11-17-2020 15";

O seguinte resultado é retornado:

+----------------------+------------+------------+-----+-------+---------------+
| topic                | partition  | offset     | key | value | pt            |
+----------------------+------------+------------+-----+-------+---------------+
| maxcompute-test-input| 0          | 0          | 1   | 1     | 11-17-2020 15 |
+----------------------+------------+------------+-----+-------+---------------+

Se a consulta não retornar linhas, verifique se consultou a partição correta indicada por show partitions, revise os detalhes de consumo do Group que assina o tópico source e cheque o tópico de dados de erro e a fila de mensagens mortas do conector em busca de mensagens que falharam na gravação.

ApsaraMQ for Kafka:Criar um conector sink do MaxCompute

Pré-requisitos

Observações de uso

Procedimento

Criar uma função RAM

Adicionar permissões

Criar os tópicos necessários para um conector sink do MaxCompute

Criar o Group necessário para um conector sink do MaxCompute

Criar e implantar um conector sink do MaxCompute

Enviar uma mensagem de teste

Visualizar dados da tabela