Elasticsearch:Migrasi data dari Elasticsearch yang dikelola sendiri ke Alibaba Cloud Elasticsearch menggunakan reindex

Skrip Batch (Python 2.7): Gunakan skrip ini untuk mereplikasi struktur indeks (pemetaan/shard) dari kluster lama ke yang baru.

#!/usr/bin/python
# -*- coding: UTF-8 -*-
# Nama file: indiceCreate.py
import sys
import base64
import time
import httplib
import json
## Host kluster Elasticsearch yang dikelola sendiri.
oldClusterHost = "old-cluster.com"
## Username untuk kluster Elasticsearch yang dikelola sendiri. Bisa dikosongkan.
oldClusterUserName = "old-username"
## Password untuk kluster Elasticsearch yang dikelola sendiri. Bisa dikosongkan.
oldClusterPassword = "old-password"
## Host kluster Alibaba Cloud Elasticsearch. Anda bisa mendapatkannya dari halaman Informasi Dasar instans Alibaba Cloud Elasticsearch.
newClusterHost = "new-cluster.com"
## Username untuk kluster Alibaba Cloud Elasticsearch.
newClusterUser = "elastic"
## Password untuk kluster 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 + "  Pengaturan aslinya adalah sebagai berikut:\n" + indexSettings
    settingsDict = json.loads(indexSettings)
    ## Secara default, jumlah shard sama dengan indeks di kluster Elasticsearch yang dikelola sendiri.
    number_of_shards = settingsDict[index]["settings"]["index"]["number_of_shards"]
    ## Secara default, jumlah replica adalah 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 + " Pemetaan aslinya adalah sebagai berikut:\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 "Pengaturan dan pemetaan untuk indeks baru " + newIndexName + " adalah sebagai berikut:\n" + createstatement
    endpoint = "/" + newIndexName
    createResult = httpPut(newClusterHost, endpoint, createstatement, newClusterUser, newClusterPassword)
    print "Hasil pembuatan indeks baru " + 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 + " kemungkinan merupakan indeks sistem dan tidak akan dibuat ulang. Tangani secara terpisah jika diperlukan."

Skenario A: Migrasi sederhana (data kecil)

Jalankan skrip ini sekali untuk setiap indeks.

#!/bin/bash
# file:reindex.sh
indexName="Nama indeks Anda"
newClusterUser="Username untuk kluster Alibaba Cloud Elasticsearch"
newClusterPass="Password untuk kluster Alibaba Cloud Elasticsearch"
newClusterHost="Host kluster Alibaba Cloud Elasticsearch"
oldClusterUser="Username untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Password untuk kluster Elasticsearch yang dikelola sendiri"
# Host kluster Elasticsearch yang dikelola sendiri harus dalam format [scheme]://[host]:[port], misalnya, http://10.37.*.*:9200.
oldClusterHost="Host kluster Elasticsearch yang dikelola sendiri"
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}'"
    }
}'

Skenario B: Migrasi inkremental (Data besar dengan timestamp)

Jika Anda memiliki bidang update_time, gunakan skrip loop ini untuk migrasi data secara bertahap. Ini meminimalkan downtime dengan memungkinkan Anda menyinkronkan "delta" (perubahan baru) setelah pemindahan awal secara massal.

#!/bin/bash
# file: circleReindex.sh
# MENGONTROL STARTUP:
# Ini adalah skrip untuk reindex jarak jauh. Persyaratan:
# 1. Indeks telah dibuat di kluster Alibaba Cloud Elasticsearch, atau kluster mendukung pembuatan otomatis dan pemetaan dinamis.
# 2. Daftar putih alamat IP harus dikonfigurasi di file YML kluster Alibaba Cloud Elasticsearch, misalnya, reindex.remote.whitelist: 172.16.**.**:9200.
# 3. Host harus dalam format [scheme]://[host]:[port].
PENGGUNAAN="Penggunaan: sh circleReindex.sh <jumlah>
       jumlah: Jumlah eksekusi. Angka negatif menunjukkan loop untuk eksekusi inkremental. Angka positif menunjukkan eksekusi satu kali atau beberapa kali.
Contoh:
        sh circleReindex.sh 1
        sh circleReindex.sh 5
        sh circleReindex.sh -1"
indexName="Nama indeks Anda"
newClusterUser="Username untuk kluster Alibaba Cloud Elasticsearch"
newClusterPass="Password untuk kluster Alibaba Cloud Elasticsearch"
oldClusterUser="Username untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Password untuk kluster Elasticsearch yang dikelola sendiri"
## http://myescluster.com
newClusterHost="Host kluster Alibaba Cloud Elasticsearch"
# Host kluster Elasticsearch yang dikelola sendiri harus dalam format [scheme]://[host]:[port], misalnya, http://10.37.*.*:9200.
oldClusterHost="Host kluster Elasticsearch yang dikelola sendiri"
timeField="Bidang waktu pembaruan"
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 ke-${reindexTimes}. Batas waktu pembaruan untuk eksekusi ini adalah ${lastTimestamp}. Hasil: ${ret}"
    if [[ ${ret} == *error* ]]; then
        hasError=true
        echo "Terjadi pengecualian selama eksekusi ini. Operasi selanjutnya dihentikan. Harap periksa."
    fi
}
function start() {
    ## Jika angka negatif, loop berjalan terus-menerus.
    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 "$PENGGUNAAN"
    exit 1
fi
echo "Mulai operasi reindex untuk indeks ${indexName}."
start $1
echo "Total ${reindexTimes} operasi reindex telah dilakukan."

Skenario C: Migrasi inkremental (Data besar tanpa timestamp)

Modifikasi kode layanan hulu untuk menambahkan bidang update_time. Setelah menambahkan bidang tersebut, pertama-tama migrasi data historis. Kemudian, gunakan metode migrasi scroll yang dijelaskan di Skenario B: Migrasi inkremental (Data besar dengan timestamp).

#!/bin/bash
# file:miss.sh
indexName="Nama indeks Anda"
newClusterUser="Username untuk kluster Alibaba Cloud Elasticsearch"
newClusterPass="Password untuk kluster Alibaba Cloud Elasticsearch"
newClusterHost="Host kluster Alibaba Cloud Elasticsearch"
oldClusterUser="Username untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Password untuk kluster Elasticsearch yang dikelola sendiri"
# Host kluster Elasticsearch yang dikelola sendiri harus dalam format [scheme]://[host]:[port], misalnya, http://10.37.*.*:9200
oldClusterHost="Host kluster Elasticsearch yang dikelola sendiri"
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}'"
    }
}'