ApsaraMQ for Kafka:Create an Elasticsearch sink connector

ApsaraMQ for Kafka

• Activer la fonctionnalité de connecteur pour votre instance.

• Créer une rubrique à utiliser comme source de données.

Function Compute

• Activer Function Compute.

Elasticsearch

• Créer un cluster et un index Elasticsearch dans la console Elasticsearch. Utilisez la version 7.0 ou ultérieure pour assurer la compatibilité avec le client Function Compute (version 7.7.0).

• Ajouter le bloc CIDR du endpoint Function Compute à la liste d'autorisation Elasticsearch. Pour les tests initiaux, spécifiez 0.0.0.0/0 afin d'autoriser toutes les adresses IP du VPC, puis restreignez la plage après avoir vérifié la connectivité.

Informations à recueillir

Rassemblez les informations suivantes avant de démarrer l'assistant.

Étape 1 : Configurer les informations de base

À l'étape Configure Basic Information, définissez le nom du connecteur et examinez les détails de l'instance.

Cliquez sur Next.

Étape 2 : Configurer le service source

À l'étape Configure Source Service, sélectionnez Message Queue for Apache Kafka comme service source et configurez les paramètres suivants.

Cliquez sur Configure Runtime Environment pour afficher les paramètres supplémentaires.

Étape 3 : Configurer le service de destination

À l'étape Configure Destination Service, sélectionnez Elasticsearch comme service de destination et configurez les paramètres suivants.

Cliquez sur Create.

Déployer le connecteur

Après la création, le connecteur apparaît sur la page Connectors. Cliquez sur Deploy dans la colonne Actions pour démarrer le connecteur.

Envoyer un message de test

1. Sur la page Connectors, recherchez le connecteur et cliquez sur Test dans la colonne Actions.

2. Dans le panneau Send Message, définissez l'option Method of Sending sur Console.

3. Dans le champ Message Key, saisissez une clé, par exemple demo.

4. Dans le champ Message Content, saisissez un corps JSON, par exemple :

      {"key": "test"}

5. Pour l'option Send to Specified Partition, cliquez sur Yes et saisissez un Partition ID (par exemple, 0) pour cibler une partition spécifique, ou cliquez sur No pour laisser le système en attribuer une. Pour rechercher les ID de partition, consultez la section Afficher l'état des partitions.

Vous pouvez également envoyer des messages de test via Docker ou un SDK. Sélectionnez l'option correspondante dans le champ Method of Sending et suivez les instructions affichées à l'écran.

Vérifier l'index Elasticsearch

1. Connectez-vous à la console Kibana.

2. Exécutez la requête suivante pour rechercher dans l'index cible :

      GET /<index_name>/_search

3. Confirmez que la réponse contient le message que vous avez envoyé. Une réponse réussie ressemble à ceci :

      {
        "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
              }
            }
          ]
        }
      }