Elasticsearch:Migrate data from a self-managed Elasticsearch cluster to Alibaba Cloud Elasticsearch by using reindex

Data migration using reindex is supported only for single-zone instances. If you use a multi-zone instance, use one of the following solutions to migrate data from your self-managed ES cluster to Alibaba Cloud:

• If the source data volume is large, use the OSS snapshot method. For more information, see Advanced: Migrate a self-managed ES cluster to Alibaba Cloud ES using OSS.

• To filter the source data, use the Logstash migration solution. For more information, see Migrate data from a self-managed Elasticsearch cluster to Alibaba Cloud using Logstash.

You have performed the following operations:

• Create a single-zone Alibaba Cloud ES instance.

  For more information, see Create an Alibaba Cloud Elasticsearch instance.

• Prepare a self-managed ES cluster and the data to be migrated.

  If you do not have a self-managed ES cluster, you can use an Alibaba Cloud ECS instance to build one. For more information, see Install and run Elasticsearch. The self-managed ES cluster must meet the following conditions:

  • The network type of the ECS instance where the cluster is located must be a virtual private cloud (VPC). ECS instances connected using ClassicLink are not supported. The ECS instance must be in the same VPC as the Alibaba Cloud ES instance.

  • The security group of the ECS instance must not restrict access from the IP addresses of the nodes in the Alibaba Cloud ES instance. You can view the node IP addresses in the Kibana console. Port 9200 must also be open.

  • The self-managed cluster must be able to connect to the Alibaba Cloud ES instance. On the machine from which you run the script, run the curl -XGET http://<host>:9200 command to verify the connection.

    Note

    You can run the scripts in this topic on any machine that can access port 9200 of both the self-managed ES cluster and the Alibaba Cloud ES cluster.

Alibaba Cloud ES provides two deployment modes: basic control (v2) architecture and cloud-native new control (v3) architecture. You can check the deployment mode in the Basic Information section of the instance.

For clusters that use the cloud-native new control (v3) architecture, you must use PrivateLink to establish a private connection to the Alibaba Cloud ES cluster for cross-cluster reindex operations. Select a solution from the following table based on your business scenario.

The following example shows a Python 2 script that you can use to create indexes in batches in the Alibaba Cloud ES cluster. These indexes must have the same configurations as the indexes that you want to migrate from the self-managed ES cluster. By default, the number of replicas for the new indexes is set to 0.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# File name: indiceCreate.py
import sys
import base64
import time
import httplib
import json
## The host of the self-managed Elasticsearch cluster.
oldClusterHost = "old-cluster.com"
## The username for the self-managed Elasticsearch cluster. This can be empty.
oldClusterUserName = "old-username"
## The password for the self-managed Elasticsearch cluster. This can be empty.
oldClusterPassword = "old-password"
## The host of the Alibaba Cloud Elasticsearch cluster. You can obtain it from the Basic Information page of the Alibaba Cloud Elasticsearch instance.
newClusterHost = "new-cluster.com"
## The username for the Alibaba Cloud Elasticsearch cluster.
newClusterUser = "elastic"
## The password for the Alibaba Cloud Elasticsearch cluster.
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 are as follows:\n" + indexSettings
    settingsDict = json.loads(indexSettings)
    ## By default, the number of shards is the same as that of the index in the self-managed Elasticsearch cluster.
    number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
    ## By default, the 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 mapping is as follows:\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 "The settings and mapping for the new index " + newIndexName + " are as follows:\n" + createstatement
    endpoint = "/" + newIndexName
    createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
    print "Result of creating the new index " + 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 + " might be a system index and will not be re-created. Handle it separately if needed."

Small data volume

You can use the following script.

#!/bin/bash
# file:reindex.sh
indexName="Your index name"
newClusterUser="Username for the Alibaba Cloud Elasticsearch cluster"
newClusterPass="Password for the Alibaba Cloud Elasticsearch cluster"
newClusterHost="Host of the Alibaba Cloud Elasticsearch cluster"
oldClusterUser="Username for the self-managed Elasticsearch cluster"
oldClusterPass="Password for the self-managed Elasticsearch cluster"
# The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port], for example, http://10.37.*.*:9200.
oldClusterHost="Host of the self-managed Elasticsearch cluster"
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}'"
    }
}'

Large data volume, no delete operations, and with update time

If the data volume is large and no delete operations are performed, you can use scroll migration to reduce the downtime of write services. Scroll migration requires a field, such as an update time field, to track new data. After the initial data migration is complete, stop all write operations. Then, run the reindex operation one last time to migrate the latest updates. Finally, switch your read and write operations to the Alibaba Cloud ES cluster.

#!/bin/bash
# file: circleReindex.sh
# CONTROLLING STARTUP:
# This is a script for remote reindexing. Requirements:
# 1. The index has been created in the Alibaba Cloud Elasticsearch cluster, or the cluster supports automatic creation and dynamic mapping.
# 2. An IP address whitelist must be configured in the YML file of the Alibaba Cloud Elasticsearch cluster, for example, reindex.remote.whitelist: 172.16.**.**:9200.
# 3. The host must be in the format of [scheme]://[host]:[port].
USAGE="Usage: sh circleReindex.sh <count>
       count: The number of executions. A negative number indicates a loop for incremental execution. A positive number indicates a one-time or multiple executions.
Example:
        sh circleReindex.sh 1
        sh circleReindex.sh 5
        sh circleReindex.sh -1"
indexName="Your index name"
newClusterUser="Username for the Alibaba Cloud Elasticsearch cluster"
newClusterPass="Password for the Alibaba Cloud Elasticsearch cluster"
oldClusterUser="Username for the self-managed Elasticsearch cluster"
oldClusterPass="Password for the self-managed Elasticsearch cluster"
## http://myescluster.com
newClusterHost="Host of the Alibaba Cloud Elasticsearch cluster"
# The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port], for example, http://10.37.*.*:9200.
oldClusterHost="Host of the self-managed Elasticsearch cluster"
timeField="Update time field"
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 "The ${reindexTimes}th reindex. The update deadline for this execution is ${lastTimestamp}. Result: ${ret}"
    if [[ ${ret} == *error* ]]; then
        hasError=true
        echo "An exception occurred during this execution. Subsequent operations are interrupted. Please check."
    fi
}
function start() {
    ## If the number is negative, the loop runs continuously.
    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 "Start the reindex operation for the index ${indexName}."
start $1
echo "A total of ${reindexTimes} reindex operations were performed."

Large data volume, no delete operations, and without update time

If the data volume is large and the index mapping does not contain an update time field, you must modify the upstream service code to add this field. After you add the field, first migrate the historical data. Then, use the scroll migration method described in the previous section.

#!/bin/bash
# file:miss.sh
indexName="Your index name"
newClusterUser="Username for the Alibaba Cloud Elasticsearch cluster"
newClusterPass="Password for the Alibaba Cloud Elasticsearch cluster"
newClusterHost="Host of the Alibaba Cloud Elasticsearch cluster"
oldClusterUser="Username for the self-managed Elasticsearch cluster"
oldClusterPass="Password for the self-managed Elasticsearch cluster"
# The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port], for example, http://10.37.*.*:9200
oldClusterHost="Host of the self-managed Elasticsearch cluster"
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}'"
    }
}'