ApsaraMQ for Kafka:Crie um conector sink do AnalyticDB

Pré-requisitos

Para mais informações, consulte Prerequisites.

Etapa 1: Crie os recursos de destino

Crie um recurso do AnalyticDB for MySQL ou do AnalyticDB for PostgreSQL.

• AnalyticDB for MySQL: No console do AnalyticDB for MySQL, crie um cluster e uma conta de banco de dados, conecte-se ao cluster e crie um banco de dados. Para mais informações, consulte Crie um cluster, Crie uma conta de banco de dados, Connect to a cluster e Crie um banco de dados.

• AnalyticDB for PostgreSQL: No console do AnalyticDB for PostgreSQL, crie uma instância e uma conta de banco de dados e faça logon no banco de dados. Para mais informações, consulte Crie uma instância, Crie e gerencie usuários e Client connection.

Este exemplo usa um banco de dados do AnalyticDB for MySQL chamado adb_sink_database e uma tabela chamada adb_sink_table.

Etapa 2: Crie e ative o conector sink do AnalyticDB

1. Faça logon no console do ApsaraMQ for Kafka. 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.

   • Configure a tarefa

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

        Parâmetro	Descrição	Exemplo
        Region	A região da instância de origem do Message Queue for Apache Kafka.	China (Beijing)
        Kafka instance	A instância de origem do 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): Cria automaticamente um group ID no formato GID_EVENTBRIDGE_xxx. Use Existing: Selecione um group ID existente. Não compartilhe este group ID com outros serviços para evitar interromper o 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 apenas se a Network configuration estiver definida como Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Obrigatório apenas se a Network configuration estiver definida como Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        Security group	Obrigatório apenas se a Network configuration estiver definida como Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	O formato de codificação para o conteúdo da mensagem. Recomendamos o formato Json caso não haja requisitos específicos de codificação. Json: Codifica dados binários como um objeto JSON no payload em 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 codificada em Base64 no payload.	Json
        Messages	Parâmetro de Advanced configuration. O número máximo de mensagens por lote. O sistema envia uma solicitação 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, para agregação e envio das mensagens ao destino. Valores válidos: 0 a 15. O valor 0 indica entrega imediata das mensagens.	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 executar 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 o Service Type como AnalyticDB, configure os parâmetros a seguir e clique em Save.

        Parâmetro	Descrição	Exemplo
        Instance type	Selecione o tipo de banco de dados da instância de destino. AnalyticDB for MySQL AnalyticDB for PostgreSQL	MySQL version
        AnalyticDB instance ID	Selecione a instância de destino.	gp-bp10uo5n536wd****
        Database name	Selecione o banco de dados de destino.	adb_sink_database
        Table name	Selecione a tabela de destino.	adb_sink_table
        Data Mapping	Use expressões JSONPath para definir regras de extração de dados. Quando o Data Format está definido como Json na etapa Source, o ApsaraMQ for Kafka encapsula os dados transmitidos em uma estrutura CloudEvents, conforme o exemplo abaixo: { "data": { "topic": "demo-topic", "partition": 0, "offset": 2, "timestamp": 1739756629123, "headers": { "headers": [], "isReadOnly": false }, "key":"adb-sink-k1", "value": { "userid":"xiaoming", "source":"shanghai" } }, "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" }	Mapeie cada coluna da tabela de destino para um campo na mensagem de origem usando uma expressão JSONPath. Por exemplo, para mapear o campo userid da mensagem para uma coluna da tabela, use a expressão $.data.value.userid.
        Database username	Insira o nome de usuário da conta de banco de dados.	user
        Database password	Insira a senha da conta de banco de dados.
        Network configuration	VPC: Conecte-se ao AnalyticDB por meio de uma VPC. Public Network: Conecte-se ao AnalyticDB pela internet pública.	VPC
        VPC	Selecione o ID da VPC. Obrigatório apenas quando a Network configuration está definida como VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Selecione o ID do vSwitch. Obrigatório apenas quando a Network configuration está definida como VPC. Importante Após selecionar um vSwitch, adicione o bloco CIDR do vSwitch à lista de permissões de endereços IP da instância do AnalyticDB for MySQL. Para mais informações, consulte Configure an IP address whitelist.	vsw-bp1gbjhj53hdjdkg****
        Security group	Selecione o grupo de segurança. Obrigatório apenas quando a Network configuration está definida como VPC.	test_group

4. Retorne à página Tasks. Encontre a tarefa criada e clique em Enable na coluna Actions.

5. Na caixa de diálogo Note, leia a mensagem e clique em OK.

   A tarefa leva de 30 a 60 segundos para iniciar. Acompanhe o progresso na coluna Status da página Tasks.

Etapa 3: Verifique o conector sink do AnalyticDB

1. Na página Tasks, encontre a tarefa e clique no nome do tópico de origem na coluna Event Source.

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

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

   Nota

   O corpo da mensagem deve estar no formato JSON. O sistema extrai e grava os campos especificados nas regras de mapeamento de dados nas colunas correspondentes da tabela de destino.

   Na caixa de diálogo Start to Send and Consume Message, selecione a aba Console. Defina a Message key como adb-sink-k1 e o Message body como {"userid":"xiaoming","source":"shanghai"}. Para Send to a specific partition, selecione No e clique em OK.

4. Na página Tasks, encontre a tarefa e clique no nome da instância de destino na coluna Event Target.

5. Na página Basic Information da instância, clique em Log On to Database no canto superior direito.

6. No console do Data Management Service (DMS), execute a seguinte instrução SQL para consultar todos os dados da tabela.

   SELECT * FROM  adb_sink_table;

   A consulta deve retornar um registro com userid igual a xiaoming e source igual a shanghai. Isso confirma a gravação bem-sucedida dos dados na tabela de destino.