すべてのプロダクト
Search

このプロダクト

    Elasticsearch:Logstash を使用した Azure Event Hubs から Alibaba Cloud Elasticsearch へのデータ同期

    ドキュメントセンター

    Elasticsearch:Logstash を使用した Azure Event Hubs から Alibaba Cloud Elasticsearch へのデータ同期

    最終更新日:Jun 24, 2026

    Alibaba Cloud Logstash パイプラインを使用して、Azure Event Hubs から Alibaba Cloud Elasticsearch インスタンスにデータを同期できます。

    操作手順

    1. ステップ 1：環境とインスタンスの準備
    2. ステップ 2：Logstash パイプラインの作成と設定
    3. ステップ 3：結果の検証

    ステップ 1：環境とインスタンスの準備

    1. Alibaba Cloud Elasticsearch インスタンスを作成し、インデックスの自動作成を有効にします。この例では、バージョン 7.10 のインスタンスを使用します。
      詳細については、「Alibaba Cloud Elasticsearch インスタンスの作成」および「YML パラメーターの設定」をご参照ください。
    2. Alibaba Cloud Logstash インスタンスを作成し、パブリックネットワークアクセスのために NAT Gateway を設定します。この例では、バージョン 7.4 のインスタンスを使用します。
      詳細については、「Alibaba Cloud Logstash インスタンスの作成」をご参照ください。
      Alibaba Cloud Logstash インスタンスは VPC にデプロイされます。Logstash がパブリックインターネット経由で Azure Event Hubs と通信できるようにするには、VPC 用に NAT Gateway を設定します。詳細については、「NAT Gateway を使用したパブリックネットワークアクセスの設定」をご参照ください。
      説明 自己管理の Logstash をインストールする場合は、Alibaba Cloud Elasticsearch インスタンスと同じ VPC 内に ECS インスタンスを購入し、EIP を割り当てます。要件を満たす ECS インスタンスが既にある場合は、新しく購入する必要はありません。
    3. Azure Event Hubs 環境を準備します。
      詳細については、Azure Event Hubs の公式ドキュメントをご参照ください。

    ステップ 2：Logstash パイプラインの作成と設定

    1. [Logstash クラスター] ページに移動します。

    2. 対象のクラスターに移動します。

      1. 上部のナビゲーションバーで、クラスターが存在するリージョンを選択します。

      2. [Logstash クラスター] ページで、対象のクラスターを見つけてその ID をクリックします。

    3. 左側のナビゲーションウィンドウで、[パイプライン管理] をクリックします。

    4. [パイプラインの作成] をクリックします。

    5. [タスクの作成] ページで、[パイプライン ID] を入力し、パイプラインを設定します。
      この例では、次のパイプライン設定を使用します。
      input {
  azure_event_hubs {
     event_hub_connections => ["Endpoint=sb://abc-****.****.cn/;SharedAccessKeyName=gem-****-es-consumer;SharedAccessKey=******;EntityPath=xxxxxx"]
     initial_position => "beginning"
     threads => 2
     decorate_events => true
     consumer_group => "group-kl"
     storage_connection => "DefaultEndpointsProtocol=https;AccountName=xxxxx;AccountKey=*******;EndpointSuffix=core.****.cn"
     storage_container => "lettie_container"
   }
}
filter {

}
output {
  elasticsearch {
    hosts => ["es-cn-tl****5r50005adob.elasticsearch.aliyuncs.com:9200"]
    index => "test-log"
    password => "xxxxxx"
    user => "elastic"
  }
}
      表 1. Input パラメーター
      パラメーター 説明
      event_hub_connections 読み取り元のイベントハブの接続文字列のリスト。接続文字列には、イベントハブの `EntityPath` が含まれます。詳細については、「event_hub_connections」をご参照ください。
      説明 各イベントハブには、個別の event_hub_connections パラメーターが必要です。他のパラメーターは、すべてのイベントハブに適用されます。
      initial_position イベントハブ内でデータの読み取りを開始する位置。有効な値：`beginning` (デフォルト)、`end`、`look_back`。詳細については、「initial position 」をご参照ください。
      threads イベント処理用のスレッドの総数。詳細については、「threads」をご参照ください。
      decorate_events イベントハブ名、コンシューマーグループ、プロセッサホスト、パーティション、オフセット、シーケンス、タイムスタンプ、イベントサイズなど、イベントハブのメタデータを同期データに含めるかどうかを指定します。詳細については、「decorate events」をご参照ください。
      consumer_group イベントハブからデータを読み取るために使用されるコンシューマーグループ。Logstash 専用のコンシューマーグループを作成し、すべての Logstash ノードがそれを使用して作業を調整するようにしてください。詳細については、「consumer group」をご参照ください。
      storage_connection Azure Blob Storage アカウントの接続文字列。このストレージは再起動後もオフセットを永続化するため、複数の Logstash ノードが異なるパーティションを処理できます。このパラメーターが設定されている場合、Logstash は再起動後に停止した場所から処理を再開します。設定されていない場合は、initial_position で指定された位置から再開します。詳細については、「storage connection」をご参照ください。
      storage_container オフセットを永続化し、複数の Logstash ノード間の作業を調整するために使用されるストレージコンテナの名前。詳細については、「storage container」をご参照ください。
      説明 オフセットの上書きを避けるため、パイプラインごとに一意の storage_container 名を使用してください。同じデータを複数のサービスに書き込む場合は、パイプラインごとに異なる名前を指定します。
      表 2. Output パラメーター
      パラメーター 説明
      hosts ご利用の Alibaba Cloud Elasticsearch インスタンスのエンドポイント。フォーマットは http://<ご利用の Alibaba Cloud Elasticsearch インスタンス ID>.elasticsearch.aliyuncs.com:9200 です。
      index 送信先インデックスの名前。
      user Elasticsearch インスタンスにアクセスするためのユーザー名。デフォルト値：`elastic`。
      password 指定されたユーザーのパスワード。`elastic` ユーザーのパスワードは、Elasticsearch インスタンスの作成時に設定します。パスワードを忘れた場合は、リセットできます。詳細については、「インスタンスのアクセスパスワードのリセット」をご参照ください。

      パイプライン設定の詳細については、「Logstash 設定ファイル」をご参照ください。

    6. [Next step] をクリックし、パイプラインのパラメーターを設定します。

      パラメーター

      説明

      Pipeline Workers

      フィルターおよび出力ステージのワーカースレッド数。イベントがバックログされている場合や CPU が十分に活用されていない場合は、この値を増やします。デフォルト：CPU コア数。

      Pipeline Batch Size

      ワーカーがフィルターと出力を実行する前に収集する最大イベント数。バッチサイズを大きくするとメモリ使用量が増加し、より大きな JVM ヒープサイズ (LS_HEAP_SIZE) が必要になる場合があります。デフォルト：125。

      Pipeline Batch Delay

      サイズが小さいバッチをワーカースレッドにディスパッチするまでの待機時間 (ミリ秒)。デフォルト：50 ms。

      Queue Type

      イベントバッファリングのための内部キューイングモデル。有効な値：

      • [MEMORY]：デフォルト。メモリ内キューを使用します。

      • PERSISTED：ディスクベースの永続キュー。

      Queue Max Bytes

      キューが保存できるデータの最大量 (MB)。値は 1 から 2<sup>53</sup>-1 までの整数である必要があります。デフォルト値：1024。

      説明

      この値が合計ディスク容量未満であることを確認してください。

      Queue Checkpoint Writes

      チェックポイントが強制される前に書き込まれる最大イベント数 (永続キューのみ)。0 は制限なしを意味します。デフォルト：1024。
      警告

      保存とデプロイを行うと、インスタンスが再起動されます。ビジネスに影響がない場合にのみ続行してください。

    7. [Save] または [保存とデプロイ] をクリックします。

      • [Save]：パイプラインの設定を保存しますが、適用はしません。保存後、[パイプライン管理] ページに戻ります。[パイプラインリスト] セクションの [Actions] 列で [デプロイ] をクリックすると、インスタンスが再起動され、設定が適用されます。

      • [保存とデプロイ]：設定を保存してデプロイし、インスタンスを再起動して変更を適用します。

    ステップ 3：結果の検証

    1. Elasticsearch クラスターの Kibana コンソールにログインし、Kibana のホームページに移動します。

    2. 左側のナビゲーションメニューで、[開発ツール] をクリックします。

    3. [コンソール] で、次のコマンドを実行して同期されたデータを確認します。 
      GET test-log3/_search
{
  "query":{
    "match":{
      "message":"L23"
     }
   }
}
      期待される出力：预期结果