ApsaraMQ for Kafka:Elasticsearch sink コネクタの作成

ステップ 1：基本情報の構成

[基本情報の構成] ステップで、コネクタ名を設定し、インスタンスの詳細を確認します。

[次へ] をクリックします。

ステップ 2：ソースサービスの構成

[ソースサービスの構成] ステップで、ソースサービスとして [Message Queue for Apache Kafka] を選択し、以下のパラメーターを設定します。

[ランタイム環境の構成] をクリックして、追加のパラメーターを展開します。

ステップ 3：宛先サービスの構成

[宛先サービスの構成] ステップで、宛先サービスとして [Elasticsearch] を選択し、以下のパラメーターを設定します。

[作成] をクリックします。

コネクタのデプロイ

作成後、コネクタは [コネクタ] ページに表示されます。[操作] 列の [デプロイ] をクリックしてコネクタを開始します。

テストメッセージの送信

1. [コネクタ] ページでコネクタを見つけ、[操作] 列の [テスト] をクリックします。

2. [メッセージの送信] パネルで、[送信方法] を [コンソール] に設定します。

3. [メッセージキー] フィールドに、キー (例：demo) を入力します。

4. [メッセージ内容] フィールドに、JSON 本文 (例：) を入力します。

      {"key": "test"}

5. 「[指定されたパーティションに送信]」の場合、「[はい]」をクリックし、特定のパーティションを対象とするために[パーティション ID]（例： 0）を入力するか、または「[いいえ]」をクリックしてシステムに割り当てさせます。パーティション ID を確認するには、「パーティションのステータスを表示」をご参照ください。

テストメッセージは [Docker] または [SDK] を通じても送信できます。[送信方法] フィールドで対応するオプションを選択し、画面の指示に従ってください。

Elasticsearch インデックスの確認

1. Kibana コンソールにログインします。

2. 次のクエリを実行して、ターゲットインデックスを検索します。

      GET /<index_name>/_search

3. 応答に送信したメッセージが含まれていることを確認します。成功した応答は次のようになります。

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