How to create a MaxCompute sink connector - ApsaraMQ for Kafka: Central de Documentação do Alibaba Cloud

Pré-requisitos

Para obter as etapas detalhadas, consulte Prerequisites for sink connectors.

Observações

Para usar o recurso de particionamento do MaxCompute, crie uma coluna de partição adicional chamada time com o tipo de dados STRING ao criar a tabela.

Etapa 1: Crie o recurso de destino

Crie uma tabela usando o cliente do MaxCompute. Para obter mais informações, consulte Create a table.

O exemplo a seguir cria uma tabela chamada kafka_to_maxcompute com três colunas e o recurso de particionamento ativado:

CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT) PARTITIONED by (time STRING);

Caso não use o recurso de particionamento, use a seguinte instrução:

CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT);

Após a execução da instrução, o sistema retorna o seguinte resultado:

Na página Tables, visualize as informações da tabela kafka_to_maxcompute criada. O tipo de partição é partitioned table e o tipo de tabela é internal table. O esquema da tabela contém três campos: topic (string), valueName (string) e valueAge (bigint). Nenhum deles é chave primária. O campo de partição é time (string).

Etapa 2: Crie e inicie o conector

Acesse o console do ApsaraMQ for Kafka. Na página Overview, selecione uma região na seção Resource Distribution.
No painel de navegação à esquerda, escolha Connector Ecosystem Integration > Tasks.
Na página Tasks, clique em Create Task.

Na página Create Task, configure os parâmetros Task Name e Description. Em seguida, siga as instruções na tela para configurar os demais parâmetros.

Criação da tarefa

No assistente de configuração Source, defina Data Provider como ApsaraMQ for Kafka, configure os parâmetros a seguir e clique em Next.

Parâmetro	Descrição	Exemplo
Region	A região onde a instância do ApsaraMQ for Kafka está localizada.	China (Hangzhou)
ApsaraMQ for Kafka Instance	O ID da instância do ApsaraMQ for Kafka na qual os dados que você deseja rotear são produzidos.	alikafka_post-cn-9hdsbdhd****
Topic	O tópico na instância do ApsaraMQ for Kafka no qual os dados que você deseja rotear são produzidos.	guide-sink-topic
Group ID	O ID do grupo na instância do ApsaraMQ for Kafka no qual os dados que você deseja rotear são produzidos. Quickly Create: O sistema cria automaticamente um grupo cujo ID está no formato GID_EVENTBRIDGE_xxx. Use Existing Group: selecione o ID de um grupo existente que não esteja em uso. Se você selecionar um grupo existente em uso, isso afetará a publicação e a assinatura das mensagens existentes.	Use Existing Group
Consumer Offset	Latest Offset: As mensagens são consumidas a partir do offset mais recente. Earliest Offset: As mensagens são consumidas a partir do offset mais antigo.	Latest Offset
Network Configuration	Se for necessária a transmissão de dados entre fronteiras, selecione Self-managed Internet. Caso contrário, selecione Basic Network.	Basic Network
Data Format	O recurso de formato de dados codifica dados binários entregues pela source em um formato específico. Há suporte para vários formatos de dados. Se não houver requisitos especiais de codificação, especifique Json como valor. Json: Codifica dados binários no formato JSON com base em UTF-8 e os insere no payload. Text: O formato padrão. Codifica dados binários em uma string com base em UTF-8 e a insere no payload. Binary: Codifica dados binários em strings com base na codificação Base64 e, em seguida, os insere no payload.	Text
Messages	Parâmetro de Advanced Configuration. O número máximo de mensagens a serem enviadas em um lote. Uma solicitação é enviada somente quando o número de mensagens acumuladas atinge esse valor. O valor deve ser um número inteiro de 1 a 10.000.	2000
Interval (Unit: Seconds)	Parâmetro de Advanced Configuration. O intervalo no qual a função é chamada. O sistema agrega mensagens e as envia para o function compute nesse intervalo. O valor deve ser um número inteiro de 0 a 15. A unidade é segundos. Um valor de 0 significa que as mensagens são entregues imediatamente.	3

Na etapa Filtering, defina um padrão de evento para filtrar solicitações. Para obter mais informações, consulte Event patterns.
Na etapa Transformation, configure a limpeza de dados para processamentos complexos, como divisão, mapeamento, enriquecimento e roteamento dinâmico. Para obter mais informações, consulte Use function compute to clean message data.

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

Parâmetro	Descrição	Exemplo
AccessKey ID	O AccessKey ID da sua conta Alibaba Cloud para acessar o MaxCompute.	yourAccessKeyID
AccessKey Secret	O AccessKey Secret da sua conta Alibaba Cloud.	yourAccessKeySecret
MaxCompute Project Name	selecione um projeto existente do MaxCompute.	test_compute
MaxCompute Table Name	selecione uma tabela existente do MaxCompute.	kafka_to_maxcompute
MaxCompute Table Input Parameter	Após selecionar uma tabela, configure uma value extraction rule para cada coluna. No exemplo de mensagem a seguir, o valor da coluna `topic` é extraído do campo `topic`, portanto, a value extraction rule é `$.topic`. `{ 'data': { 'topic': 't_test', 'partition': 2, 'offset': 1, 'timestamp': 1717048990499, 'headers': { 'headers': [], 'isReadOnly': False }, 'key': 'MaxCompute-K1', 'value': 'MaxCompute-V1' }, 'id': '9b05fc19-9838-4990-bb49-ddb942307d3f-2-1', 'source': 'acs:alikafka', 'specversion': '1.0', 'type': 'alikafka:Topic:Message', 'datacontenttype': 'application/json; charset=utf-8', 'time': '2024-05-30T06:03:10.499Z', 'aliyunaccountid': '1413397765616316' }`	topic: `$.data.topic` valuename: `$.data.value` valueage: `$.data.offset`
Partition Dimension	Close: Desativa o recurso de particionamento. Enable: Ativa o recurso de particionamento. Se você ativar o particionamento, configure parâmetros como o valor da partição: O valor da partição aceita as variáveis de tempo {yyyy}, {MM}, {dd}, {HH} e {mm}, que representam ano, mês, dia, hora e minuto, respectivamente. As variáveis de tempo diferenciam maiúsculas de minúsculas. O valor da partição também pode ser uma constante.	Enable {yyyy}-{MM}-{dd}.{HH}:{mm}.suffix
Network configuration	VPC: Entregue mensagens do Kafka ao MaxCompute por meio de uma VPC. Public Network: Entregue mensagens do Kafka ao MaxCompute pela internet.	internet
VPC	selecione um ID de VPC. Este parâmetro é obrigatório apenas se você defina Network Configuration como VPC.	vpc-bp17fapfdj0dwzjkd****
vSwitch	selecione um ID de vSwitch. Este parâmetro é obrigatório apenas se você defina Network Configuration como VPC.	vsw-bp1gbjhj53hdjdkg****
Security Group	selecione um grupo de segurança. Este parâmetro é obrigatório apenas se você defina Network Configuration como VPC.	test_group

Propriedades da tarefa

configure políticas de nova tentativa e filas de mensagens mortas para lidar com erros de entrega. Para obter mais informações, consulte Retry policies and dead-letter queues.

Após concluir as configurações anteriores, salve a configuração e clique em Save. Na página Tasks, localize a tarefa do conector de destino do MaxCompute criada. A coluna Status exibe Starting. Quando o status mudar para Running, o processo de criação estará concluído.

Etapa 3: Teste o conector

Na página Tasks, localize o conector de destino do MaxCompute e clique em tópico source na coluna Event Source.
Na página de detalhes do tópico, clique em Send Test Message.
No painel Start to Send and Consume Message, configure o conteúdo da mensagem da seguinte forma e clique em OK.

Na guia Console, defina Message key como MaxCompute-K1, defina Message body como MaxCompute-V1 e defina Send to specified partition como No.

Acesse o console do MaxCompute e execute a seguinte instrução SQL e visualize as informações da partição.

show PARTITIONS kafka_to_maxcompute;

O sistema retorna o seguinte resultado:

OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
time=2024-05-31.16:37.suffix
OK
2024-05-31 16:42:49 INFO ==================================================================
2024-05-31 16:42:49 INFO Exit code of the Shell command 0
2024-05-31 16:42:49 INFO --- Invocation of Shell command completed ---
2024-05-31 16:42:49 INFO Shell run successfully!
2024-05-31 16:42:49 INFO Current task status: FINISH
2024-05-31 16:42:49 INFO Cost time is: 1.411s

Com base nas informações da partição, execute a seguinte instrução e visualize os dados na partição.
```
SELECT * FROM kafka_to_maxcompute WHERE time="2024-05-31.16:37.suffix";
```
A consulta retorna um registro: topic é xxx (mascarado), valueName é MaxCompute-V1, valueAge é 4 e time é 2024-05-31.16:37.suffix. Isso confirma a exportação dos dados do ApsaraMQ for Kafka para a tabela particionada do MaxCompute.