ApsaraMQ for Kafka:Criar um conector de sink do MaxCompute

Criar uma função do RAM

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

1. Faça login no console do RAM.

2. Na barra de navegação à esquerda, escolha Identities > Roles.

3. Na página Roles, clique em Create Role.

   A figura a seguir mostra o botão Create Role na página Roles.

   

4. No painel Create Role, execute as seguintes operações.

5. Defina o tipo de entidade confiável como Alibaba Cloud Service e clique em Next.

6. 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.

7. Na página Roles, localize AliyunKafkaMaxComputeUser1 e clique em AliyunKafkaMaxComputeUser1.

8. Na página AliyunKafkaMaxComputeUser1, clique na aba Trust Policy e depois em Edit Trust Policy.

9. 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.

Adicionar permissões

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

Para mais informações sobre as permissões acima 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 do RAM como um usuário, portanto, cada comando nesta seção tem como alvo o objeto RAM$<accountid>:role/aliyunkafkamaxcomputeuser1.

1. Faça login no cliente do MaxCompute.

2. Execute o seguinte comando para adicionar a função do RAM como usuário.

   add user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

3. Conceda à função do RAM o privilégio mínimo necessário para acessar o MaxCompute.

4. Execute o seguinte comando para conceder à função do RAM as permissões no projeto.

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

2. Execute o seguinte comando para conceder à função do 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 de sink do MaxCompute

No console do ApsaraMQ for Kafka, é possível criar manualmente os cinco tópicos exigidos por um conector de 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.

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 tópicos, 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.

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.

1. 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.

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

Criar os grupos de consumidores necessários para um conector de sink do MaxCompute

No console do ApsaraMQ for Kafka, crie manualmente o Group utilizado pela tarefa de sincronização de dados do conector de 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}}.

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 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 o nome do Group no campo Group ID, adicione uma breve descrição no campo Description, adicione tags e clique em OK.

7. Após a criação, o Group aparecerá na lista da página Groups. Insira o nome dele no campo Connector Consumer Group durante a criação do conector.

Criar e implantar um conector de sink do MaxCompute

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

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.

1. Na página do assistente de configuração Create Connector, execute as seguintes operações.

   1. 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. Convenções de nomenclatura: - Pode conter dígitos, letras minúsculas e hifens (-), mas não pode começar com hífen. Deve ter entre 1 e 48 caracteres. - Precisa ser único na mesma instância do ApsaraMQ for Kafka. A tarefa de sincronização de dados do conector deve usar um Group chamado connect-nome da tarefa. Se você não criar o Group manualmente, o sistema o fará automaticamente.	kafka-maxcompute-sink
      Instance	Por padrão, o nome e o ID da instância são especificados.	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 um tópico e um grupo de consumidores, escolha a criação manual de recursos e insira as informações dos seus recursos existentes. Caso contrário, opte pela criação automática.

      Parâmetro	Descrição	Exemplo
      Data Source Topic	Tópico de origem cujos dados você deseja sincronizar.	maxcompute-test-input
      Consumer Thread Concurrency	Número de threads de consumo simultâneas para o tópico de origem. Valor padrão: 6. Valores válidos: - 1 - 2 - 3 - 6 - 12	6
      Consumer Offset	Offset a partir do qual as mensagens são consumidas. 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 do 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 do ApsaraMQ for Kafka. Por padrão, utiliza-se o vSwitch especificado durante a implantação da instância.	vsw-bp1d2jgg81***
      Failure Handling Policy	Define se o sistema deve continuar 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 imprime o log de erro. - Stop Subscription: Interrompe a assinatura da partição com erro e imprime 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 contatar o suporte técnico.	Continue Subscription
      Resource Creation Method	Método usado para criar os tópicos e o Group necessários ao conector. Clique em Configure Runtime Environment para exibir este parâmetro. Valores válidos: - Auto: O sistema cria os tópicos e o Group necessários. Selecione esta opção se não os tiver criado previamente. - Manual: Selecione esta opção se já tiver criado os tópicos e o Group e insira os nomes nos campos a seguir.	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 ser 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: Deve ser armazenamento local. - cleanup.policy: A política de limpeza de logs deve ser compact.	connect-offset-kafka-maxcompute-sink
      Task Configuration Topic	Tópico que armazena as 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: Deve ser 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 armazena 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: Recomenda-se definir como 6. - Mecanismo de armazenamento: Deve ser 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 que armazena dados anômalos do framework Connect. Clique em Configure Runtime Environment para exibir este parâmetro. Para economizar 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: Recomenda-se definir como 6. - Mecanismo de armazenamento: Pode ser armazenamento local ou cloud storage.	connect-error-kafka-maxcompute-sink
      Error Data Topic	Tópico que armazena dados anômalos do sink. Clique em Configure Runtime Environment para exibir este parâmetro. Para economizar 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: Recomenda-se definir como 6. - Mecanismo de armazenamento: Pode ser armazenamento local ou cloud storage.	connect-error-kafka-maxcompute-sink

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

   Nota

   Se a instância à qual o conector pertence estiver na região China (Hangzhou) ou China (Chengdu), uma caixa de diálogo Service Authorization aparecerá para cada uma das funções vinculadas ao serviço 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 abaixo e clique em Create. Se as funções vinculadas ao serviço já existirem, elas não serão recriadas e a caixa de diálogo Service Authorization não aparecerá.

   Parâmetro	Descrição	Exemplo
   Endpoint	Endpoint de serviço do 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 se 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 estiverem em regiões diferentes. Para usar 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 do MaxCompute.	test_kafka
   Region for Table	Região onde a tabela do MaxCompute reside.	China (Hangzhou)
   Alibaba Cloud Account ID	ID da conta Alibaba Cloud do MaxCompute.	188***
   RAM Role	Nome da função do 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: Apenas a chave da mensagem é mantida e gravada na coluna key da tabela do MaxCompute. - VALUE: Apenas o valor da mensagem é mantido e gravado na coluna value da tabela. - DEFAULT: Tanto a chave quanto o valor são mantidos e gravados nas respectivas colunas. > 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. 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: Mantém apenas a chave. A string da chave é separada por vírgulas e as partes são gravadas na tabela seguindo a ordem dos índices. - Modo VALUE: Mantém apenas o valor. A string do valor é separada por vírgulas e as partes são gravadas na tabela seguindo a ordem dos índices.	TEXT
   Partition	Granularidade da partição. Valor padrão: HOUR. Valores válidos: - DAY: Dados gravados em nova partição diariamente. - HOUR: Dados gravados em nova partição a cada hora. - MINUTE: Dados gravados em 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 de origem. Valor padrão: GMT+08:00.	GMT+08:00

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

2. 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 de sink do MaxCompute, envie uma mensagem para o tópico de origem do ApsaraMQ for Kafka para testar a sincronização dos dados com o MaxCompute.

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

1. 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 para 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 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 de origem do ApsaraMQ for Kafka, visualize os dados da tabela no cliente do MaxCompute para verificar o recebimento.

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.

1. Faça login no cliente do MaxCompute.

2. Execute o seguinte comando para visualizar as partições de dados da tabela.

   show partitions test_kafka;

   O seguinte resultado é retornado:

   pt=11-17-2020 15
   
   OK

3. Execute o seguinte comando 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 de origem e verifique o tópico de dados de erro e a fila de mensagens mortas do conector em busca de mensagens que falharam na gravação.