ApsaraMQ for Kafka:Create an Elasticsearch sink connector

ApsaraMQ for Kafka

• Ative o recurso de conector da sua instância.

• Crie um tópico para usar como fonte de dados.

Function Compute

• Ative o Function Compute.

Elasticsearch

• Crie um cluster e um índice do Elasticsearch no console do Elasticsearch. Use a versão 7.0 ou posterior para garantir compatibilidade com o cliente do Function Compute (versão 7.7.0).

• Adicione o bloco CIDR do endpoint do Function Compute à lista de permissões do Elasticsearch. Para testes iniciais, especifique 0.0.0.0/0 para permitir todos os endereços IP na VPC; depois, restrinja o intervalo após verificar a conectividade.

Informações a coletar

Reúna os detalhes abaixo antes de iniciar o assistente.

Etapa 1: Configurar informações básicas

Na etapa Configure Basic Information, defina o nome do conector e revise os detalhes da instância.

Clique em Next.

Etapa 2: Configurar o serviço de origem

Na etapa Configure Source Service, selecione Message Queue for Apache Kafka como serviço de origem e configure os parâmetros a seguir.

Clique em Configure Runtime Environment para expandir parâmetros adicionais.

Etapa 3: Configurar o serviço de destino

Na etapa Configure Destination Service, selecione Elasticsearch como serviço de destino e configure os parâmetros a seguir.

Clique em Create.

Implantar o conector

Após a criação, o conector aparece na página Connectors. Clique em Deploy na coluna Actions para iniciar o conector.

Enviar uma mensagem de teste

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

2. No painel Send Message, defina Method of Sending como Console.

3. No campo Message Key, insira uma chave, por exemplo, demo.

4. No campo Message Content, insira um corpo JSON, por exemplo:

      {"key": "test"}

5. Para Send to Specified Partition, clique em Yes e insira um Partition ID (por exemplo, 0) para direcionar a uma partição específica, ou clique em No para deixar o sistema atribuir uma. Para consultar IDs de partição, veja Visualizar status da partição.

Também é possível enviar mensagens de teste via Docker ou um SDK. Selecione a opção correspondente no campo Method of Sending e siga as instruções na tela.

Verificar o índice do Elasticsearch

1. Faça login no console do Kibana.

2. Execute a consulta a seguir para pesquisar no índice de destino:

      GET /<index_name>/_search

3. Confirme se a resposta contém a mensagem enviada. Uma resposta bem-sucedida tem a seguinte aparência:

      {
        "took": 8,
        "timed_out": false,
        "_shards": {
          "total": 5,
          "successful": 5,
          "skipped": 0,
          "failed": 0
        },
        "hits": {
          "total": {
            "value": 1,
            "relation": "eq"
          },
          "max_score": 1.0,
          "hits": [
            {
              "_index": "product_****",
              "_type": "_doc",
              "_id": "TX3TZHgBfHNEDGoZ****",
              "_score": 1.0,
              "_source": {
                "msg_body": {
                  "key": "test",
                  "offset": 2,
                  "overflowFlag": false,
                  "partition": 2,
                  "timestamp": 1616599282417,
                  "topic": "dv****",
                  "value": "test1",
                  "valueSize": 8
                },
                "doc_as_upsert": true
              }
            }
          ]
        }
      }