自建Elasticsearch迁移

1. 创建索引。
2. 数据迁移。

• 自建Elasticsearch所在的ECS必须是VPC网络（不支持Classiclink方式打通的ECS），并且自建Elasticsearch必须与阿里云Elasticsearch在同一个VPC下。
• 您可以通过中控机器（或者任意一台机器）执行文档中的脚本，前提是该中控机器可以同时访问新旧Elasticsearch集群的9200端口。
• 自建Elasticsearch所在的ECS的VPC安全组不能限制IP白名单，并且需要开启9200端口。
• 自建Elasticsearch所在的ECS的VPC安全组不能限制阿里云Elasticsearch实例的各节点IP（Kibana控制台可查看阿里云Elasticsearch实例各节点的IP）。
• 自建Elasticsearch与阿里云Elasticsearch实例已经连通。可以在执行脚本的机器上使用curl -XGET http://<host>:9200进行验证。

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 文件名：indiceCreate.py
import sys
import base64
import time
import httplib
import json
## 老集群host（ip+port）
oldClusterHost = "old-cluster.com"
## 老集群用户名，可为空
oldClusterUserName = "old-username"
## 老集群密码，可为空
oldClusterPassword = "old-password"
## 新集群host（ip+port）
newClusterHost = "new-cluster.com"
## 新集群用户名，可为空
newClusterUser = "new-username"
## 新集群密码，可为空
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 + "  原始settings如下：\n" + indexSettings
    settingsDict = json.loads(indexSettings)
    ## 分片数默认和老集群索引保持一致
    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 + " 原始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 "新索引 " + newIndexName + " 的setting和mapping如下：\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 + "  或许是系统索引，不会重新创建，如有需要，请单独处理～"

• 问题：执行curl命令时提示{"error":"Content-Type header [application/x-www-form-urlencoded] is not supported","status":406}。
  解决方法：可以在curl命令中添加-H "Content-Type: application/json"参数重试。

  // 获取老集群中所有索引信息，如果没有权限可去掉"-u user:pass"参数，oldClusterHost为老集群的host，注意替换。
    curl -u user:pass -XGET http://oldClusterHost/_cat/indices | awk '{print $3}'
    // 参考上面返回的索引列表，获取需要迁移的指定用户索引的setting和mapping，注意替换indexName为要查询的用户索引名。
    curl -u user:pass -XGET http://oldClusterHost/indexName/_settings,_mapping?pretty=true
    // 参考上面获取到的对应索引的_settings,_mapping信息，在新集群中创建对应索引，索引副本数可以先设置为0，用于加快数据同步速度，数据迁移完成后再重置副本数为1。
    //其中newClusterHost是新集群的host，testindex是已经创建的索引名，testtype是对应索引的type。
    curl -u user:pass -XPUT http://<newClusterHost>/<testindex> -d '{
      "testindex" : {
          "settings" : {
              "number_of_shards" : "5", //假设老集群中对应索引的shard数是5个
              "number_of_replicas" : "0" //设置索引副本为0
            }
          },
          "mappings" : { //假设老集群中对应索引的mappings配置如下
              "testtype" : {
                  "properties" : {
                      "uid" : {
                          "type" : "long"
                      },
                      "name" : {
                          "type" : "text"
                      },
                      "create_time" : {
                        "type" : "long"
                      }
                  }
             }
         }
     }
  }'

• 问题：数据同步速度太慢。
  解决方法： 如果单索引数据量比较大，可以在迁移前将目的索引的副本数设置为 0，刷新时间为 -1。待数据迁移完成后，再更改回来，这样可以加快数据同步速度。

  // 迁移索引数据前可以先将索引副本数设为0，不刷新，用于加快数据迁移速度。
  curl -u user:password -XPUT 'http://<host:port>/indexName/_settings' -d' {
          "number_of_replicas" : 0,
          "refresh_interval" : "-1"
  }'
  // 索引数据迁移完成后，可以重置索引副本数为1，刷新时间1s（1s是默认值）。
  curl -u user:password -XPUT 'http://<host:port>/indexName/_settings' -d' {
          "number_of_replicas" : 1,
          "refresh_interval" : "1s"
  }'