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

Migrasi data menggunakan reindex hanya didukung untuk instance zona tunggal. Jika Anda menggunakan instance multi-zona, gunakan salah satu solusi berikut untuk memigrasikan data dari kluster ES yang dikelola sendiri ke Alibaba Cloud:

• Jika volume data sumber besar, gunakan metode snapshot OSS. Untuk informasi selengkapnya, lihat Tingkat Lanjut: Migrasi kluster ES yang dikelola sendiri ke ES Alibaba Cloud menggunakan OSS.

• Untuk memfilter data sumber, gunakan solusi migrasi Logstash. Untuk informasi selengkapnya, lihat Migrasi data dari kluster Elasticsearch yang dikelola sendiri ke Alibaba Cloud menggunakan Logstash.

Anda telah melakukan operasi berikut:

• Buat instance Alibaba Cloud ES zona tunggal.

  Untuk informasi selengkapnya, lihat Buat instance Alibaba Cloud Elasticsearch.

• Persiapkan kluster ES yang dikelola sendiri dan data yang akan dimigrasikan.

  Jika Anda belum memiliki kluster ES yang dikelola sendiri, Anda dapat menggunakan instance ECS Alibaba Cloud untuk membangunnya. Untuk informasi selengkapnya, lihat Instal dan jalankan Elasticsearch. Kluster ES yang dikelola sendiri harus memenuhi kondisi berikut:

  • Jenis jaringan instance ECS tempat kluster berada harus berupa virtual private cloud (VPC). Instance ECS yang terhubung menggunakan ClassicLink tidak didukung. Instance ECS harus berada dalam VPC yang sama dengan instance ES Alibaba Cloud.

  • Grup keamanan instance ECS tidak boleh membatasi akses dari alamat IP node di instance ES Alibaba Cloud. Anda dapat melihat alamat IP node di konsol Kibana. Port 9200 juga harus terbuka.

  • Kluster yang dikelola sendiri harus dapat terhubung ke instance ES Alibaba Cloud. Pada mesin tempat Anda menjalankan skrip, jalankan perintah curl -XGET http://<host>:9200 untuk memverifikasi koneksi.

    Catatan

    Anda dapat menjalankan skrip dalam topik ini pada mesin apa pun yang dapat mengakses port 9200 baik kluster ES yang dikelola sendiri maupun kluster ES Alibaba Cloud.

Alibaba Cloud ES menyediakan dua mode penyebaran: arsitektur kontrol dasar (v2) dan arsitektur kontrol baru cloud-native (v3). Anda dapat memeriksa mode penyebaran di bagian Informasi Dasar instance.

Untuk kluster yang menggunakan arsitektur kontrol baru cloud-native (v3), Anda harus menggunakan PrivateLink untuk membuat koneksi privat ke kluster ES Alibaba Cloud demi operasi reindex antar kluster. Pilih solusi dari tabel berikut berdasarkan skenario bisnis Anda.

Contoh berikut menunjukkan skrip Python 2 yang dapat Anda gunakan untuk membuat indeks secara batch di kluster ES Alibaba Cloud. Indeks ini harus memiliki konfigurasi yang sama dengan indeks yang ingin Anda migrasikan dari kluster ES yang dikelola sendiri. Secara default, jumlah replika untuk indeks baru diatur ke 0.

#!/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"
## Nama pengguna untuk kluster Elasticsearch yang dikelola sendiri. Bisa kosong.
oldClusterUserName = "old-username"
## Kata sandi untuk kluster Elasticsearch yang dikelola sendiri. Bisa kosong.
oldClusterPassword = "old-password"
## Host kluster Elasticsearch Alibaba Cloud. Anda dapat memperolehnya dari halaman Informasi Dasar instance Elasticsearch Alibaba Cloud.
newClusterHost = "new-cluster.com"
## Nama pengguna untuk kluster Elasticsearch Alibaba Cloud.
newClusterUser = "elastic"
## Kata sandi untuk kluster Elasticsearch Alibaba Cloud.
newClusterPassword = "new-password"
DEFAULT_REPLICAS = 0
def httpRequest(method, host, endpoint, params="", username="", password=""):
    conn = httplib.HTTPConnection(host)
    headers = {}
    if (username != "") :
        'Halo {name}, usia Anda adalah {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 replika 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."

Volume data kecil

Anda dapat menggunakan skrip berikut.

#!/bin/bash
# file:reindex.sh
indexName="Nama indeks Anda"
newClusterUser="Nama pengguna untuk kluster Elasticsearch Alibaba Cloud"
newClusterPass="Kata sandi untuk kluster Elasticsearch Alibaba Cloud"
newClusterHost="Host kluster Elasticsearch Alibaba Cloud"
oldClusterUser="Nama pengguna untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Kata sandi 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}'"
    }
}'

Volume data besar, tanpa operasi hapus, dan dengan waktu pembaruan

Jika volume data besar dan tidak ada operasi hapus, Anda dapat menggunakan migrasi scroll untuk mengurangi waktu henti layanan tulis. Migrasi scroll memerlukan bidang, seperti bidang waktu pembaruan, untuk melacak data baru. Setelah migrasi data awal selesai, hentikan semua operasi tulis. Kemudian, jalankan operasi reindex sekali lagi untuk memigrasikan pembaruan terbaru. Terakhir, alihkan operasi baca dan tulis Anda ke kluster ES Alibaba Cloud.

#!/bin/bash
# file: circleReindex.sh
# MENGONTROL STARTUP:
# Ini adalah skrip untuk reindexing jarak jauh. Persyaratan:
# 1. Indeks telah dibuat di kluster Elasticsearch Alibaba Cloud, atau kluster mendukung pembuatan otomatis dan pemetaan dinamis.
# 2. Daftar putih alamat IP harus dikonfigurasi di file YML kluster Elasticsearch Alibaba Cloud, 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="Nama pengguna untuk kluster Elasticsearch Alibaba Cloud"
newClusterPass="Kata sandi untuk kluster Elasticsearch Alibaba Cloud"
oldClusterUser="Nama pengguna untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Kata sandi untuk kluster Elasticsearch yang dikelola sendiri"
## http://myescluster.com
newClusterHost="Host kluster Elasticsearch Alibaba Cloud"
# 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. Silakan periksa."
    fi
}
function start() {
    ## Jika angkanya 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."

Volume data besar, tanpa operasi hapus, dan tanpa waktu pembaruan

Jika volume data besar dan pemetaan indeks tidak berisi bidang waktu pembaruan, Anda harus memodifikasi kode layanan hulu untuk menambahkan bidang ini. Setelah menambahkan bidang tersebut, pertama-tama migrasikan data historis. Kemudian, gunakan metode migrasi scroll yang dijelaskan di bagian sebelumnya.

#!/bin/bash
# file:miss.sh
indexName="Nama indeks Anda"
newClusterUser="Nama pengguna untuk kluster Elasticsearch Alibaba Cloud"
newClusterPass="Kata sandi untuk kluster Elasticsearch Alibaba Cloud"
newClusterHost="Host kluster Elasticsearch Alibaba Cloud"
oldClusterUser="Nama pengguna untuk kluster Elasticsearch yang dikelola sendiri"
oldClusterPass="Kata sandi 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}'"
    }
}'