Elasticsearch:Use the reindex API to migrate data from a self-managed ES cluster to Alibaba Cloud

You can use the reindex API to migrate data only to single-zone instances. If you have a multi-zone instance, use one of the following solutions to migrate data from your self-managed ES cluster to Alibaba Cloud:

• If the self-managed Elasticsearch cluster stores a large volume of data, use snapshots stored in Object Storage Service (OSS). For more information, see Use OSS to migrate data from a self-managed Elasticsearch cluster to an Alibaba Cloud Elasticsearch cluster.

• If you want to filter source data, use Logstash. For more information, see Use Alibaba Cloud Logstash to migrate data from a self-managed Elasticsearch cluster to an Alibaba Cloud Elasticsearch cluster.

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 migrate.

  We recommend that you use a self-managed Elasticsearch cluster deployed on Alibaba Cloud ECS instances. For more information about how to deploy a self-managed Elasticsearch cluster, see Installing and Running Elasticsearch. The self-managed Elasticsearch cluster must meet the following requirements:

  • The ECS instances that host the cluster must be in a virtual private cloud (VPC). The ECS instances cannot be connected through ClassicLink. The VPC must be the same one that the Alibaba Cloud ES instance uses.

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

  • The cluster must be able to connect to the Alibaba Cloud ES instance. You can run the curl -XGET http://<host>:9200 command on the machine where you run the script to verify the connection.

    Note

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

Alibaba Cloud ES provides two deployment modes: Basic Management (v2) architecture and Cloud-native New Management (v3) architecture. You can identify the deployment mode of your instance on the Basic Information page.

For clusters that use the Cloud-native New Management (v3) architecture, you must use PrivateLink to establish a private connection for cross-cluster reindexing. The following table describes the solutions for different business scenarios.

The following example provides a Python 2 script to create multiple indexes from the self-managed ES cluster in the Alibaba Cloud ES cluster. By default, the number of replicas for the newly created indexes is 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 this from the Basic Information page of the instance in the Alibaba Cloud Elasticsearch console. 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 + " Original settings:\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"] ## 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 + " Original mapping:\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 of 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. It will not be recreated. Handle it separately if needed."

Small data volume

Use the following script.

#!/bin/bash # file:reindex.sh indexName="your_index_name" newClusterUser="alibaba_cloud_es_cluster_username" newClusterPass="alibaba_cloud_es_cluster_password" newClusterHost="alibaba_cloud_es_cluster_host" oldClusterUser="self_managed_es_cluster_username" oldClusterPass="self_managed_es_cluster_password" # The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port]. For example, http://10.37.*.*:9200. oldClusterHost="self_managed_es_cluster_host" 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 timestamps

If you have a large volume of data and no delete operations, you can use a rolling migration to reduce the downtime of write services. A rolling migration requires a field, such as an update timestamp, to represent the write sequence of new data. After the data migration is complete, stop the business write operations. After reindex quickly performs an update using the most recent update time, switch the read and write operations to the Alibaba Cloud ES cluster.

#!/bin/bash # file: circleReindex.sh # CONTROLLING STARTUP: # This script remotely rebuilds an index using the reindex operation. Requirements: # 1. The index is already created in the Alibaba Cloud Elasticsearch cluster, or the cluster supports automatic index 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 execution. Example: sh circleReindex.sh 1 sh circleReindex.sh 5 sh circleReindex.sh -1" indexName="your_index_name" newClusterUser="alibaba_cloud_es_cluster_username" newClusterPass="alibaba_cloud_es_cluster_password" oldClusterUser="self_managed_es_cluster_username" oldClusterPass="self_managed_es_cluster_password" ## http://myescluster.com newClusterHost="alibaba_cloud_es_cluster_host" # The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port]. For example, http://10.37.*.*:9200. oldClusterHost="self_managed_es_cluster_host" 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 "Reindex attempt #${reindexTimes}. Update deadline for this attempt: ${lastTimestamp}. Result: ${ret}" if [[ ${ret} == *error* ]]; then hasError=true echo "An error occurred during this execution. Subsequent operations are interrupted. Please check." fi } function start() { ## A negative number indicates a continuous loop. 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 "Starting 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 timestamps

If you have a large volume of data and the index mapping does not define an update timestamp field, you must modify the code of the upstream service to add the field. After you add the field, you can first migrate the historical data and then use the second method described in this topic.

#!/bin/bash # file:miss.sh indexName="your_index_name" newClusterUser="alibaba_cloud_es_cluster_username" newClusterPass="alibaba_cloud_es_cluster_password" newClusterHost="alibaba_cloud_es_cluster_host" oldClusterUser="self_managed_es_cluster_username" oldClusterPass="self_managed_es_cluster_password" # The host of the self-managed Elasticsearch cluster must be in the format of [scheme]://[host]:[port]. For example, http://10.37.*.*:9200 oldClusterHost="self_managed_es_cluster_host" 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}'" } }'