ApsaraMQ for Kafka:Criar um Elasticsearch sink connector

Pré-requisitos

Para mais informações, consulte Prerequisites.

Etapa 1: Criar recursos do service de destino

• Crie uma instância e um índice no Elasticsearch console. Para mais informações, consulte Getting started.

• Adicione o bloco CIDR do service Function Compute à lista de permissões de sua instância do Elasticsearch. Esse procedimento é necessário apenas para conexões via VPC. Para mais informações, consulte Configure a public or private IP address whitelist for an instance.

Etapa 2: Criar um Elasticsearch 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, configure os parâmetros Task Name e Description. Em seguida, siga as instruções na tela para configure os demais parâmetros.

   • Configure a task

     1. 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 segue o formato GID_EVENTBRIDGE_xxx. Use Existing Group: Selecione o ID de um grupo existente que não esteja em uso. Selecionar um grupo que já esteja em uso afeta 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	Para transmissões de dados entre regiões, selecione Self-managed Internet. Nos demais casos, selecione Basic Network.	Basic Network
        Data Format	O recurso de formato de dados codifica dados binários provenientes do source em um formato específico. Há suporte para múltiplos formatos. Caso não haja 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: Formato padrão. Codifica dados binários em uma string com base em UTF-8 e os insere no payload. Binary: Os dados binários são codificados em strings com base em Base64 e inseridos no payload.	Text
        Messages	Parâmetro de Advanced Configuration. Define o número máximo de mensagens a enviar em um lote. A requisição é enviada somente quando o número de mensagens acumuladas atinge esse valor. O valor deve ser um número inteiro entre 1 e 10.000.	2000
        Interval (Unit: Seconds)	Parâmetro de Advanced Configuration. Define o intervalo em que a função é chamada. O sistema agrega as mensagens e as envia ao Function Compute nesse intervalo. O valor deve ser um número inteiro entre 0 e 15, em segundos. O valor 0 indica entrega imediata das mensagens.	3

     2. Na etapa Filtering, defina um padrão de evento para filtrar requisições. Para mais informações, consulte Event patterns.

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

     4. Na etapa Sink, selecione Alibaba Cloud Elasticsearch acs.elasticSearch em Service Type e configure os parâmetros a seguir.

        Parâmetro	Descrição	Exemplo
        Elasticsearch Cluster	A instância do Elasticsearch que você criou.	es-cn-pe336j0gj001e****
        Cluster Logon Name	O nome de login da instância, cujo valor padrão é elastic.	elastic
        Instance logon password	A senha definida no momento da criação da instância.
        Index Name	O nome do índice que você criou. Para mais informações, consulte Getting started. O valor pode ser uma constante string ou uma variável JSONPath, como product_info ou $.data.key.	product_info
        Document Type	O tipo do documento de dados. O valor pode ser uma constante string ou uma variável JSONPath. Exemplos: _doc ou $.data.key. Nota Este parâmetro pode ser configurado apenas para versões do Elasticsearch anteriores à 7,0. O valor padrão é a constante _doc.	_doc
        Document	Selecione se deseja entregar o evento completo ou um evento parcial ao Elasticsearch. Para eventos parciais, é necessário configurar uma regra de extração JSONPath.	Complete Event
        Network Configuration	VPC: Entrega mensagens do Kafka ao Elasticsearch por meio de uma VPC. Public Network: Entrega mensagens do Kafka ao Elasticsearch pela internet.	Public Network
        VPC	A VPC que contém a instância do Elasticsearch. Este parâmetro é obrigatório apenas quando Network Configuration está definido como VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	O vSwitch ao qual a instância do Elasticsearch pertence. Este parâmetro é obrigatório apenas quando Network Configuration está definido como VPC.	vsw-bp1gbjhj53hdjdkg****
        Security Group	O grupo de segurança. Este parâmetro é obrigatório apenas quando Network Configuration está definido como VPC.	test_group

   • Task properties

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

5. Após concluir a configuração, clique em Save. Na página Tasks, localize a task do Elasticsearch sink connector que você criou. A coluna Status exibe Starting. Quando o status mudar para Running, o connector estará criado e pronto para uso.

Etapa 3: Testar o Elasticsearch sink connector

1. Na página Tasks, clique em o tópico source na coluna Event Source da task do Elasticsearch sink connector.

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 conforme descrito a seguir e clique em OK.

   Defina o método de envio como Console. Defina Message Key como es-sink-k1 e Message Content como {"esk1":1,"esk2":"v2"}. Em Send to Specified Partition, selecione No.

4. Faça login no Elasticsearch console e acesse a instância via Kibana. Para mais informações, consulte Getting started.

5. No console do Kibana, execute o comando a seguir para visualize o resultado da inserção de dados.

   GET /your-index-name/_search

   A consulta retorna um status 200 OK com um documento em que _index é product_info e _id é 1717558528. O campo _source contém os campos topic, partition, offset, timestamp, headers, key e value, com key definido como es-sink-k1 e value definido como {"esk1": 1, "esk2": "v2"}. Isso confirma que os dados foram gravados com sucesso no Elasticsearch.