Elasticsearch:reindex を使用して自己管理型 Elasticsearch クラスターから Alibaba Cloud Elasticsearch にデータを移行する

reindex を使用したデータ移行は、シングルゾーンインスタンスでのみサポートされています。マルチゾーンインスタンスを使用する場合は、次のいずれかのソリューションを使用して、自己管理型 ES クラスターから Alibaba Cloud にデータを移行します。

• ソースデータ量が大きい場合は、OSS スナップショットメソッドを使用します。詳細については、「高度な機能: OSS を使用して自己管理型 ES クラスターを Alibaba Cloud ES に移行する」をご参照ください。

• ソースデータをフィルター処理するには、Logstash 移行ソリューションを使用します。詳細については、「Logstash を使用して自己管理型 Elasticsearch クラスターから Alibaba Cloud にデータを移行する」をご参照ください。

次の操作を完了していること。

• シングルゾーンの Alibaba Cloud ES インスタンスを作成します。

  詳細については、「Alibaba Cloud Elasticsearch インスタンスの作成」をご参照ください。

• 自己管理型 ES クラスターと移行するデータを準備します。

  自己管理型 ES クラスターがない場合は、Alibaba Cloud ECS インスタンスを使用して構築できます。詳細については、「Elasticsearch のインストールと実行」をご参照ください。自己管理型 ES クラスターは、次の条件を満たす必要があります。

  • クラスターが配置されている ECS インスタンスのネットワークタイプは、VPC (virtual private cloud) である必要があります。ClassicLink を使用して接続された ECS インスタンスはサポートされていません。ECS インスタンスは、Alibaba Cloud ES インスタンスと同じ VPC にある必要があります。

  • ECS インスタンスのセキュリティグループは、Alibaba Cloud ES インスタンスのノードの IP アドレスからのアクセスを制限してはなりません。ノードの IP アドレスは Kibana コンソールで表示できます。ポート 9200 も開いている必要があります。

  • 自己管理型クラスターは、Alibaba Cloud ES インスタンスに接続できる必要があります。スクリプトを実行するマシンで、curl -XGET http://<host>:9200 コマンドを実行して接続を確認します。

    説明

    このトピックのスクリプトは、自己管理型 ES クラスターと Alibaba Cloud ES クラスターの両方のポート 9200 にアクセスできる任意のマシンで実行できます。

Alibaba Cloud ES は、基本制御 (v2) アーキテクチャとクラウドネイティブ新制御 (v3) アーキテクチャの 2 つのデプロイメントモードを提供します。デプロイメントモードは、インスタンスの基本情報セクションで確認できます。

クラウドネイティブ新制御 (v3) アーキテクチャを使用するクラスターの場合、クラスター間の reindex 操作には、PrivateLink を使用して Alibaba Cloud ES クラスターへのプライベート接続を確立する必要があります。ビジネスシナリオに基づいて、次の表からソリューションを選択してください。

次の例は、Alibaba Cloud ES クラスターにインデックスをバッチで作成するために使用できる Python 2 スクリプトを示しています。これらのインデックスは、自己管理型 ES クラスターから移行するインデックスと同じ構成を持つ必要があります。デフォルトでは、新しいインデックスのレプリカ数は 0 に設定されます。

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# ファイル名: indiceCreate.py
import sys
import base64
import time
import httplib
import json
## 自己管理型 Elasticsearch クラスターのホスト。
oldClusterHost = "old-cluster.com"
## 自己管理型 Elasticsearch クラスターのユーザー名。空にすることもできます。
oldClusterUserName = "old-username"
## 自己管理型 Elasticsearch クラスターのパスワード。空にすることもできます。
oldClusterPassword = "old-password"
## Alibaba Cloud Elasticsearch クラスターのホスト。Alibaba Cloud Elasticsearch インスタンスの [基本情報] ページから取得できます。
newClusterHost = "new-cluster.com"
## Alibaba Cloud Elasticsearch クラスターのユーザー名。
newClusterUser = "elastic"
## Alibaba Cloud Elasticsearch クラスターのパスワード。
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 + "  元の設定は次のとおりです:\n" + indexSettings
    settingsDict = json.loads(indexSettings)
    ## デフォルトでは、シャード数は自己管理型 Elasticsearch クラスターのインデックスのシャード数と同じです。
    number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
    ## デフォルトでは、レプリカ数は 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 + " 元のマッピングは次のとおりです:\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 "新しいインデックス " + newIndexName + " の設定とマッピングは次のとおりです:\n" + createstatement
    endpoint = "/" + newIndexName
    createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
    print "新しいインデックス " + newIndexName + " の作成結果: " + 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 + " はシステムインデックスである可能性があるため、再作成されません。必要に応じて別途処理してください。"

データ量が少ない場合

次のスクリプトを使用できます。

#!/bin/bash
# ファイル: reindex.sh
indexName="インデックス名"
newClusterUser="Alibaba Cloud Elasticsearch クラスターのユーザー名"
newClusterPass="Alibaba Cloud Elasticsearch クラスターのパスワード"
newClusterHost="Alibaba Cloud Elasticsearch クラスターのホスト"
oldClusterUser="自己管理型 Elasticsearch クラスターのユーザー名"
oldClusterPass="自己管理型 Elasticsearch クラスターのパスワード"
# 自己管理型 Elasticsearch クラスターのホストは、[scheme]://[host]:[port] の形式である必要があります (例: http://10.37.*.*:9200)。
oldClusterHost="自己管理型 Elasticsearch クラスターのホスト"
curl -u ${newClusterUser}:${newClusterPass} -XPOST "http://${newClusterHost}/_reindex?pretty" -H "Content-Type: application/json" -d'{
    "source": {
        "remote": {
            "host": "'${oldClusterHost}'",
            "username": "'${oldClusterUser}'",
            "password": "'${oldClusterPass}'"
        },
        "index": "'${indexName}'",
        "query": {
            "match_all": {}
        }
    },
    "dest": {
       "index": "'${indexName}'"
    }
}'

データ量が多く、削除操作がなく、更新時間がある場合

データ量が大きく、削除操作が実行されない場合は、scroll 移行を使用して書き込みサービスのダウンタイムを短縮できます。scroll 移行には、新しいデータを追跡するための更新時間フィールドなどのフィールドが必要です。初期データ移行が完了したら、すべての書き込み操作を停止します。次に、reindex 操作を最後に 1 回実行して、最新の更新を移行します。最後に、読み取りおよび書き込み操作を Alibaba Cloud ES クラスターに切り替えます。

#!/bin/bash
# ファイル: circleReindex.sh
# 起動の制御:
# これはリモート reindex 用のスクリプトです。要件:
# 1. Alibaba Cloud Elasticsearch クラスターにインデックスが作成されているか、クラスターが自動作成と動的マッピングをサポートしていること。
# 2. Alibaba Cloud Elasticsearch クラスターの YML ファイルに IP アドレスホワイトリストが設定されていること (例: reindex.remote.whitelist: 172.16.**.**:9200)。
# 3. ホストは [scheme]://[host]:[port] の形式であること。
USAGE="使用法: sh circleReindex.sh <count>
       count: 実行回数。負数は増分実行のループを示します。正数は 1 回または複数回の実行を示します。
例:
        sh circleReindex.sh 1
        sh circleReindex.sh 5
        sh circleReindex.sh -1"
indexName="インデックス名"
newClusterUser="Alibaba Cloud Elasticsearch クラスターのユーザー名"
newClusterPass="Alibaba Cloud Elasticsearch クラスターのパスワード"
oldClusterUser="自己管理型 Elasticsearch クラスターのユーザー名"
oldClusterPass="自己管理型 Elasticsearch クラスターのパスワード"
## http://myescluster.com
newClusterHost="Alibaba Cloud Elasticsearch クラスターのホスト"
# 自己管理型 Elasticsearch クラスターのホストは、[scheme]://[host]:[port] の形式である必要があります (例: http://10.37.*.*:9200)。
oldClusterHost="自己管理型 Elasticsearch クラスターのホスト"
timeField="更新時間フィールド"
reindexTimes=0
lastTimestamp=0
curTimestamp=`date +%s`
hasError=false
function reIndexOP() {
    reindexTimes=$[${reindexTimes} + 1]
    curTimestamp=`date +%s`
    ret=`curl -u ${newClusterUser}:${newClusterPass} -XPOST "${newClusterHost}/_reindex?pretty" -H "Content-Type: application/json" -d '{
        "source": {
            "remote": {
                "host": "'${oldClusterHost}'",
                "username": "'${oldClusterUser}'",
                "password": "'${oldClusterPass}'"
            },
            "index": "'${indexName}'",
            "query": {
                "range" : {
                    "'${timeField}'" : {
                        "gte" : '${lastTimestamp}',
                        "lt" : '${curTimestamp}'
                    }
                }
            }
        },
        "dest": {
            "index": "'${indexName}'"
        }
    }'`
    lastTimestamp=${curTimestamp}
    echo "${reindexTimes} 回目の reindex。この実行の更新期限は ${lastTimestamp} です。結果: ${ret}"
    if [[ ${ret} == *error* ]]; then
        hasError=true
        echo "この実行中に例外が発生しました。後続の操作は中断されます。確認してください。"
    fi
}
function start() {
    ## 数が負の場合、ループは継続的に実行されます。
    if [[ $1 -lt 0 ]]; then
        while :
        do
            reIndexOP
        done
    elif [[ $1 -gt 0 ]]; then
        k=0
        while [[ k -lt $1 ]] && [[ ${hasError} == false ]]; do
            reIndexOP
            let ++k
        done
    fi
}
## main 
if [ $# -lt 1 ]; then
    echo "$USAGE"
    exit 1
fi
echo "インデックス ${indexName} の reindex 操作を開始します。"
start $1
echo "合計 ${reindexTimes} 回の reindex 操作が実行されました。"

データ量が多く、削除操作がなく、更新時間がない場合

データ量が大きく、インデックスマッピングに更新時間フィールドが含まれていない場合は、アップストリームサービスのコードを変更してこのフィールドを追加する必要があります。フィールドを追加した後、まず既存データを移行します。次に、前のセクションで説明した scroll 移行メソッドを使用します。

#!/bin/bash
# ファイル: miss.sh
indexName="インデックス名"
newClusterUser="Alibaba Cloud Elasticsearch クラスターのユーザー名"
newClusterPass="Alibaba Cloud Elasticsearch クラスターのパスワード"
newClusterHost="Alibaba Cloud Elasticsearch クラスターのホスト"
oldClusterUser="自己管理型 Elasticsearch クラスターのユーザー名"
oldClusterPass="自己管理型 Elasticsearch クラスターのパスワード"
# 自己管理型 Elasticsearch クラスターのホストは、[scheme]://[host]:[port] の形式である必要があります (例: http://10.37.*.*:9200)
oldClusterHost="自己管理型 Elasticsearch クラスターのホスト"
timeField="updatetime"
curl -u ${newClusterUser}:${newClusterPass} -XPOST "http://${newClusterHost}/_reindex?pretty" -H "Content-Type: application/json" -d '{
    "source": {
        "remote": {
            "host": "'${oldClusterHost}'",
            "username": "'${oldClusterUser}'",
            "password": "'${oldClusterPass}'"
        },
        "index": "'${indexName}'",
        "query": {
            "bool": {
                "must_not": {
                    "exists": {
                        "field": "'${timeField}'"
                    }
                }
            }
        }
    },
    "dest": {
       "index": "'${indexName}'"
    }
}'