:外部 Elasticsearch インスタンスから Alibaba Cloud Elasticsearch インスタンスへのデータの移行

1. インデックスの作成
2. データの移行

• 外部 Elasticsearch インスタンスをホストする ECS インスタンスは、VPC ネットワークにデプロイされている必要があります。 ClassicLink を介して VPC ネットワークに接続された ECS インスタンスを使用することはできません。 外部 Elasticsearch インスタンスと Alibaba Cloud Elasticsearch インスタンスが、同じ VPC ネットワークに接続されていることを確認する必要があります。
• このトピックのスクリプトは、中間サーバーで実行できます。 中間サーバーがポート 9200 を介して両方の Elasticsearch インスタンスに接続されていることを確認します。
• Alibaba Cloud Elasticsearch インスタンスの IP アドレスを、外部 Elasticsearch インスタンスをホストする ECS インスタンスの VPC セキュリティグループに追加し、ポート 9200 を有効にします。
• Alibaba Cloud Elasticsearch インスタンスのノードの IP アドレスを、外部 Elasticsearch インスタンスをホストする ECS インスタンスの VPC セキュリティグループに追加します。 これにより、ノードは外部 Elasticsearch インスタンスに接続できます。 Alibaba Cloud Elasticsearch インスタンスの Kibana コンソールからノードの IP アドレスをクエリできます。
• 外部 Elasticsearch インスタンスと Alibaba Cloud Elasticsearch インスタンスは相互接続されています。 curl -XGET http://<host>:9200 コマンドを実行して、中間サーバーで接続をテストできます。

#! /usr/bin/python
# -*- coding: UTF-8 -*-
# File name: indiceCreate.py
import sys
import base64
import time
import httplib
import json
## The ECS instance that hosts the source Elasticsearch instance (IP address + Port).
oldClusterHost = "old-cluster.com"
# The username of the source Elasticsearch instance. The username field can be left empty.
oldClusterUserName = "old-username"
## The password of the source Elasticsearch instance. The password field can be left empty.
oldClusterPassword = "old-password"
## The ECS instance that hosts the destination Elasticsearch instance (IP address + Port).
newClusterHost = "new-cluster.com"
## The username of the destination Elasticsearch instance. The username field can be left empty.
newClusterUser = "new-username"
## The password of the destination Elasticsearch instance. The password field can be left empty.
newClusterPassword = "new-password"
DEFAULT_REPLICAS = 0
def httpRequest(method, host, endpoint, params="", username="", password=""):
    conn = httplib.HTTPConnection(host)
    headers = {}
    if (username ! = "") :
        'Hello {name}, your age is {age} !'.format(name = 'Tom', age = '20')
        base64string = base64.encodestring('{username}:{password}'.format(username = username, password = password)).replace('\n', '')
        headers["Authorization"] = "Basic %s" % base64string;
    if "GET" == method:
        headers["Content-Type"] = "application/x-www-form-urlencoded"
        conn.request(method=method, url=endpoint, headers=headers)
    else :
        headers["Content-Type"] = "application/json"
        conn.request(method=method, url=endpoint, body=params, headers=headers)
    response = conn.getresponse()
    res = response.read()
    return res
def httpGet(host, endpoint, username="", password=""):
    return httpRequest("GET", host, endpoint, "", username, password)
def httpPost(host, endpoint, params, username="", password=""):
    return httpRequest("POST", host, endpoint, params, username, password)
def httpPut(host, endpoint, params, username="", password=""):
    return httpRequest("PUT", host, endpoint, params, username, password)
def getIndices(host, username="", password=""):
    endpoint = "/_cat/indices"
    indicesResult = httpGet(oldClusterHost, endpoint, oldClusterUserName, oldClusterPassword)
    indicesList = indicesResult.split("\n")
    indexList = []
    for indices in indicesList:
        if (indices.find("open") > 0):
            indexList.append(indices.split()[2])
    return indexList
def getSettings(index, host, username="", password=""):
    endpoint = "/" + index + "/_settings"
    indexSettings = httpGet(host, endpoint, username, password)
    print index + "  The original settings: \n" + indexSettings
    settingsDict = json.loads(indexSettings)
    ## The number of shards equals those of the indexes on the source Elasticsearch instance by default.
    number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
    ## The default number of replicas is 0.
    number_of_replicas = DEFAULT_REPLICAS
    newSetting = "\"settings\": {\"number_of_shards\": %s, \"number_of_replicas\": %s}" % (number_of_shards, number_of_replicas)
    return newSetting
def getMapping(index, host, username="", password=""):
    endpoint = "/" + index + "/_mapping"
    indexMapping = httpGet(host, endpoint, username, password)
    print index + "The original mappings: \n" + indexMapping
    mappingDict = json.loads(indexMapping)
    mappings = json.dumps(mappingDict[index]["mappings"])
    newMapping = "\"mappings\" : " + mappings
    return newMapping
def createIndexStatement(oldIndexName):
    settingStr = getSettings(oldIndexName, oldClusterHost, oldClusterUserName, oldClusterPassword)
    mappingStr = getMapping(oldIndexName, oldClusterHost, oldClusterUserName, oldClusterPassword)
    createstatement = "{\n" + str(settingStr) + ",\n" + str(mappingStr) + "\n}"
    return createstatement
def createIndex(oldIndexName, newIndexName=""):
    if (newIndexName == "") :
        newIndexName = oldIndexName
    createstatement = createIndexStatement(oldIndexName)
    print "New index" + newIndexName + "Index settings and mappings: \n" + createstatement
    endpoint = "/" + newIndexName
    createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
    print "New index" + newIndexName + "Creation result:" + createResult
## main
indexList = getIndices(oldClusterHost, oldClusterUserName, oldClusterPassword)
systemIndex = []
for index in indexList:
    if (index.startswith(".")):
        systemIndex.append(index)
    else :
        createIndex(index, index)
if (len(systemIndex) > 0) :
    for index in systemIndex:
        print index + "It may be a system index that will not be recreated. You can manually recreate the index as needed."

• 問題：curl コマンドを実行すると、{"error":"Content-Type header [application/x-www-form-urlencoded] is not supported","status":406} が表示される。
  解決策：-H "Content-Type: application/json" パラメーターを curl コマンドに追加し、もう一度お試しください。

  // Obtain all the indexes on the source instance. If you do not have the required permissions, remove the "-u user:pass" parameter. Make sure that you have replaced oldClusterHost with the name of the ECS instance that hosts the source Elasticsearch instance.
    curl -u user:pass -XGET http://oldClusterHost/_cat/indices | awk '{print $3}'
    // Based on the returned indexes, obtain the setting and mapping of the index that you need to migrate for the specified user. Make sure that you have replaced indexName with the index name that you need to query.
    curl -u user:pass -XGET http://oldClusterHost/indexName/_settings,_mapping?pretty=true
    // Create a new index on the destination Elasticsearch instance according to the _settings and _mapping settings that you have obtained from the preceding step. You can set the number of index replicas to 0 to accelerate the data synchronization process, and change the number of replicas to one after the migration is complete.
    // ewClusterHost indicates the ECS instance that hosts the destination Elasticsearch instance, testindex indicates the name of the index that you have created, and testtype indicates the type of the index.
    curl -u user:pass -XPUT http://<newClusterHost>/<testindex> -d '{
      "testindex" : {
          "settings" : {
              "number_of_shards" : "5", //Specify the number of shards for the corresponding index on the source Elasticsearch instance, for example, 5
              "number_of_replicas" : "0" //Set the number of index replicas to zero
            }
          },
          "mappings" : { //Set the mapping for the index on the source Elasticsearch instance. For example, you can set the mapping as follows
              "testtype" : {
                  "properties" : {
                      "uid" : {
                          "type" : "long"
                      },
                      "name" : {
                          "type" : "text"
                      },
                      "create_time" : {
                        "type" : "long"
                      }
                  }
             }
         }
     }
  }'

• 問題：データ移行プロセスが遅すぎる。
  解決策：インデックスが大きすぎる場合、移行前にレプリカの数を 0 に、更新間隔を -1 に設定します。 データの移行後、レプリカと更新の設定を元の値に戻します。 これにより、同期プロセスが高速化されます。

  // You can set the number of index replicas to 0 and disable refresh to accelerate the migration process.
  curl -u user:password -XPUT 'http://<host:port>/indexName/_settings' -d' {
          "number_of_replicas" : 0,
          "refresh_interval" : "-1"
  }'
  // After the data is migrated, set the number of index replicas to 1 and the refresh interval to 1s (default value, which means 1 second).
  curl -u user:password -XPUT 'http://<host:port>/indexName/_settings' -d' {
          "number_of_replicas" : 1,
          "refresh_interval" : "1s"
  }'