セルフマネージド Elasticsearch クラスターから Alibaba Cloud Elasticsearch へのデータ移行 - Elasticsearch

Alibaba Cloud Logstash パイプラインを使用して、セルフマネージド Elasticsearch クラスターから Alibaba Cloud Elasticsearch インスタンスにデータを移行できます。

制限事項

セルフマネージド Elasticsearch クラスターをホストする ECS インスタンスは、VPC 内にデプロイされている必要があります。ClassicLink 経由で接続された ECS インスタンスはサポートされていません。
Alibaba Cloud Logstash インスタンスは VPC 内にデプロイされます。セルフマネージド Elasticsearch クラスターと Alibaba Cloud Logstash インスタンスが同一の VPC 内にある場合、直接接続できます。異なる VPC に存在する場合は、NAT Gateway を使用してパブリックインターネット経由で接続する必要があります。詳細については、「NAT Gateway を使用したパブリックネットワーク経由のデータ送信の設定」をご参照ください。
セルフマネージド Elasticsearch クラスターをホストする ECS インスタンスのセキュリティグループは、Logstash インスタンスのすべてのノード IP アドレスからのポート 9200 へのトラフィックを許可する必要があります。ノード IP アドレスは、基本情報ページで確認できます。
本トピックでは、セルフマネージド Elasticsearch 8.17 クラスター、Alibaba Cloud Elasticsearch 8.17 インスタンス、および Alibaba Cloud Logstash 8.11.4 インスタンスを例として使用しています。提供されるスクリプトはこのシナリオにのみ適用され、他の構成との互換性がない可能性があります。

ステップ 1：環境の準備

セルフマネージド Elasticsearch クラスターをセットアップします。

セルフマネージド Elasticsearch クラスターを Alibaba Cloud ECS 上にデプロイします。本トピックではバージョン 8.17 を例として使用します。詳細については、「Elasticsearch のインストールと実行」をご参照ください。
Alibaba Cloud Logstash インスタンスを作成します。

セルフマネージド Elasticsearch クラスターをホストする ECS インスタンスと同じ VPC 内に Logstash インスタンスを作成します。詳細については、「Alibaba Cloud Logstash インスタンスの作成」をご参照ください。
送信先の Alibaba Cloud Elasticsearch インスタンスを作成し、自動インデックスを有効にします。
- Logstash インスタンスと同じ VPC および同じバージョンで Elasticsearch インスタンスを作成します。本トピックではバージョン 8.17 を例として使用します。詳細については、「Alibaba Cloud Elasticsearch インスタンスの作成」をご参照ください。
- 自動インデックスの有効化方法については、「YML パラメーターの設定」をご参照ください。
  
  説明
  Logstash はデータ構造ではなくデータのみを同期します。自動インデックスを有効にすると、移行後にデータ構造が不整合になる可能性があります。一貫性を確保するには、開始前に送信先 Elasticsearch インスタンスに空のインデックスを作成してください。インデックス作成時に、ソースインデックスの mappings および settings をコピーし、適切なシャード数を設定します。

ステップ 2：Logstash パイプラインの設定と実行

Logstash クラスターページに移動します。
対象クラスターに移動します。
1. 上部ナビゲーションバーで、クラスターが配置されているリージョンを選択します。
2. Logstash クラスター ページで、対象クラスターを見つけ、その ID をクリックします。
左側のナビゲーションウィンドウで、パイプライン管理 をクリックします。
パイプラインの作成 をクリックします。

タスクの作成 ページで、パイプライン ID を入力し、パイプラインを設定します。

以下のパイプライン構成を使用します。

input {
  elasticsearch {
    hosts => ["http://<IP address of the self-managed Elasticsearch master node>:9200"]
    user => "elastic"
    index => "*,-.monitoring*,-.security*,-.kibana*"
    password => "your_password"
    docinfo => true
  }
}
filter {
}
output {
  elasticsearch {
    hosts => ["http://es-cn-mp91cbxsm000c****.elasticsearch.aliyuncs.com:9200"]
    user => "elastic"
    password => "your_password"
    index => "%{[@metadata][input][elasticsearch][_index]}"
    document_id => "%{[@metadata][input][elasticsearch][_id]}"
  }    
  file_extend {
        path => "/ssd/1/ls-cn-v0h1kzca****/logstash/logs/debug/test"
    }
}

表 1. パラメーター

パラメーター	説明
hosts	セルフマネージドまたは Alibaba Cloud Elasticsearch サービスのエンドポイントです。input セクションでは、`http://<IP address of the self-managed Elasticsearch master node>:<port>` の形式を使用します。output セクションでは、`http://<ID of the Alibaba Cloud Elasticsearch instance>.elasticsearch.aliyuncs.com:9200` の形式を使用します。重要 <IP address of the self-managed Elasticsearch master node>、<port>、および <ID of the Alibaba Cloud Elasticsearch instance> は、実際の値に置き換えてください。
user	セルフマネージドまたは Alibaba Cloud Elasticsearch サービスにアクセスするユーザー名です。重要 user パラメーターおよび password パラメーターは必須です。セルフマネージド Elasticsearch クラスターに X-Pack がインストールされていない場合は、これらのパラメーターを空のままにできます。 Alibaba Cloud Elasticsearch インスタンスにアクセスするデフォルトのユーザー名は elastic であり、本ガイドでもこれを使用しています。カスタムユーザーを使用する場合は、必要なロールおよび権限をユーザーに割り当てる必要があります。詳細については、「Elasticsearch X-Pack を使用したロールベースアクセス制御」をご参照ください。
password	セルフマネージドまたは Alibaba Cloud Elasticsearch サービスにアクセスするパスワードです。
index	同期するインデックスの名前を指定します。input パラメーターを ,-.monitoring,-.security,-.kibana に設定すると、`.` で始まるシステムインデックスを除くすべてのインデックスを同期します。output パラメーターを %{[@metadata][input][elasticsearch][_index]} に設定すると、メタデータからインデックス名を一致させます。これにより、Alibaba Cloud Elasticsearch のインデックス名がセルフマネージド Elasticsearch のインデックス名と同一になります。
docinfo	true に設定すると、`elasticsearch` input プラグインがセルフマネージド Elasticsearch ドキュメントからインデックス、タイプ、ID などのドキュメントメタデータを抽出します。
document_id	%{[@metadata][input][elasticsearch][_id]} に設定すると、メタデータからドキュメント ID を一致させます。これにより、送信先インスタンスのドキュメント ID がソースのものと一致します。
file_extend	任意です。デバッグログを有効にし、path パラメーターを使用してデバッグログの出力パスを設定します。このパラメーターの設定を推奨します。有効にすると、コンソールで直接出力を確認できます。有効にしない場合、送信先で出力を検証した後、コンソールに戻ってパイプラインを修正する必要があり、時間がかかります。詳細については、「パイプライン構成のデバッグの使用」をご参照ください。重要 file_extend パラメーターには logstash-output-file_extend プラグインが必要です。「プラグインのインストールまたはアンインストール」をご参照ください。path パラメーターのデフォルト値はシステム指定のパスです。変更しないでください。デバックログ設定の開始をクリックすることで、path を取得することもできます。

Elasticsearch input プラグインは、Alibaba Cloud Elasticsearch クラスターからドキュメントを読み取ることで、テストログの一括インポートなどのタスクに役立ちます。デフォルトでは、すべてのデータを読み取った後にこの処理は停止します。ただし、Alibaba Cloud Logstash インスタンスは継続的に実行され、停止したプロセスを自動的に再起動します。単一タスクのシナリオでは、この動作により重複データが書き込まれる可能性があります。これを防ぐには、schedule パラメーターを cron 式とともに使用して、タスクを一度だけ実行するように設定します。パイプラインはこの 1 回の実行後に停止し、繰り返しを防止します。詳細については、公式の「Logstash ドキュメント：スケジューリング」をご参照ください。

たとえば、3 月 5 日 13 時 20 分にタスクをスケジュールするには、次のようになります。

schedule => "20 13 5 3 *"

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

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 です。

警告

構成を保存してデプロイすると、インスタンスが再起動されます。ビジネスに影響がないことを確認してから進めてください。

Save または 保存とデプロイ をクリックします。
- Save：パイプライン構成を保存しますが、適用はしません。保存後、パイプライン管理 ページに戻ります。パイプラインリスト セクションで、Actions 列の デプロイ をクリックして、インスタンスを再起動し構成を適用できます。
- 保存とデプロイ：構成を保存およびデプロイし、インスタンスを再起動して変更を適用します。

ステップ 3：移行結果の確認

Alibaba Cloud Elasticsearch で、Kibana コンソールにログインします。左側のナビゲーションウィンドウで、アイコンをクリックし、Management > Dev Tools を選択します。

説明
本トピックでは Alibaba Cloud Elasticsearch 8.17 を例として使用しています。ご利用のバージョンによってユーザーインターフェースが異なる場合があります。
コンソール で、GET /_cat/indices?v コマンドを実行して、正常に移行されたインデックスを確認します。

返された結果から、orders インデックスのヘルス状態が green、ステータスが open、docs.count が 1 であることがわかり、データが正常に移行されたことを示しています。

よくある質問

Q：セルフマネージド Elasticsearch クラスターをホストする ECS インスタンスと Alibaba Cloud Logstash インスタンスが異なるアカウントに属している場合、ネットワーク接続を確立するにはどうすればよいですか？

A：ECS インスタンスと Logstash インスタンスが異なるアカウントに属している場合、それぞれ異なる VPC に存在します。Cloud Enterprise Network (CEN) を使用して 2 つの VPC を接続してください。詳細については、「ステップ 3：ネットワークインスタンスのアタッチ」をご参照ください。
Q：Logstash がデータを書き込む際にエラーが発生した場合はどうすればよいですか？

A：診断および解決手順については、「Logstash データ書き込みの問題のトラブルシューティング」をご参照ください。

Q：Alibaba Cloud Logstash 8.11 を使用して、セルフマネージド Elasticsearch 7.2 インスタンスから Alibaba Cloud Elasticsearch 7.10 インスタンスへデータを移行できますか？

A：はい。ただし、以下の重要な要件を満たす必要があります。

ネットワークおよび認証

セルフマネージド Elasticsearch → Alibaba Cloud Logstash
- セルフマネージド Elasticsearch インスタンスはパブリック IP を持ち、許可リストを設定する必要があります。
- input プラグインの認証を設定します：user/password（Basic 認証）または SSL 証明書。
Alibaba Cloud Logstash → Alibaba Cloud Elasticsearch
- 同一 VPC 内での直接プライベート接続（推奨）。この方法では、インスタンスの関連付けによるパスワードなしのアクセスが提供されます。
- インスタンスが異なる VPC にある場合は、セキュリティグループおよび許可リストを設定します。

Logstash 構成例

input {
  elasticsearch {
    hosts => ["http://<public_IP_of_self-managed_ES>:9200"]
    user => "self_es_user"
    password => "self_es_password"
    index => "source_index"
    query => '{ "query": { "match_all": {} } }'
    size => 500
    scroll => "5m"
    docinfo => true  # _source メタデータを保持し、書き込み時に _id を指定するために使用します。
  }
}
filter {
  # 任意：フィールドのクリーンアップ、マッピングの変換などを追加します。
}
output {
  elasticsearch {
    hosts => ["https://<internal_endpoint_of_Alibaba_Cloud_ES>:9200"]
    # Alibaba Cloud Logstash は通常、送信先 Elasticsearch インスタンスを選択した後に認証情報を自動的に挿入します。
    index => "target_index"
    document_id => "%{[@metadata][_id]}"  # ドキュメント ID の一貫性を確保します。
    ssl => true
    ssl_certificate_verification => false # テスト環境では無効にできます。本番環境では CA 証明書の使用を推奨します。
  }
}