如果您需要将自建Elasticsearch中的全量或增量数据迁移至阿里云Elasticsearch，可通过在ECS中自建Logstash，并通过Logstash的管道配置功能实现。本文在ECS上部署自建Elasticsearch和自建Logstash，并通过Logstash将自建Elasticsearch中的数据全量迁移至阿里云Elasticsearch。
注意事项
自建Logstash所在的ECS需要与阿里云Elasticsearch集群在同一专有网络下，同时该Logstash需要能够同时访问源Elasticsearch集群（自建）和目标Elasticsearch集群（阿里云）。
数据迁移可以全量迁移或增量迁移。如果业务侧时刻存在写入更新，首次迁移时，需先全量迁移，再通过时间标识字段（或其他可标识增量的字段）进行增量迁移，否则迁移后新数据极易被旧数据覆盖。如果已有全量数据，可以只通过标识字段实现增量数据迁移。
操作流程
开通阿里云Elasticsearch服务，在ECS服务器部署自建Elasticsearch、准备待迁移数据和部署自建Logstash。
在ECS服务器运行Python脚本迁移索引元数据。
通过Logstash管道配置功能，将自建Elasticsearch中的全量数据迁移至阿里云Elasticsearch中。
步骤一：准备环境与实例
创建阿里云Elasticsearch实例。
具体操作请参见创建阿里云Elasticsearch实例。本文使用的测试环境如下。
环境项
环境信息
地域
华东1（杭州）。
版本
通用商业版7.10.0。
实例规格配置
3个可用区、3个数据节点、单节点4核CPU、16 GB内存、100 GB ESSD云盘。
创建ECS实例，用于部署自建Elasticsearch、自建Kibana和自建Logstash。
具体操作请参见自定义购买实例。本文使用的测试环境如下。
环境项
环境信息
地域
华东1（杭州）。
实例规格
4 vCPU 16 GiB内存。
镜像
公共镜像、CentOS 7.9 64位。
存储
系统盘、ESSD云盘、100 GiB。
网络
与阿里云Elasticsearch相同的专有网络，选中分配公网IPv4地址，并按使用流量计费，带宽峰值为100 Mbps。
安全组
入方向添加5601端口（即Kibana端口），在授权对象中添加您客户端的IP地址。
重要如果您的客户端处在家庭网络或公司局域网中，您需要在授权对象中添加局域网的公网出口IP地址，而非客户端机器的IP地址。建议您通过浏览器访问cip.cc查询。
您也可以在授权对象中添加0.0.0.0/0，表示允许所有IPv4地址访问ECS实例。此配置会导致ECS实例完全暴露在公网中，增加安全风险，生产环境尽量避免。
部署自建Elasticsearch。
本文使用的自建Elasticsearch版本为7.6.2，1个数据节点，具体操作步骤如下：
连接ECS服务器。
具体操作，请参见使用Workbench登录Linux实例。
使用root用户权限创建elastic用户。
useradd elastic设置elastic用户的密码。
passwd elastic系统将提示您输入和确认elastic用户的密码。
将root用户切换为elastic用户。
su -l elastic下载Elasticsearch软件安装包并解压缩。
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.6.2-linux-x86_64.tar.gz tar -zvxf elasticsearch-7.6.2-linux-x86_64.tar.gz启动Elasticsearch。
进入Elasticsearch的安装目录下，启动服务。
cd elasticsearch-7.6.2 ./bin/elasticsearch -d验证Elasticsearch服务是否正常运行。
cd ~ curl localhost:9200正常情况下，返回结果中会显示Elasticsearch版本号和
"You Know, for Search"。[elastic@vm01 ~]$ curl localhost:9200 { "name" : "vm01", "cluster_name" : "elasticsearch", "cluster_uuid" : "SRB4pnk4SmS-YHzsrxxx", "version" : { "number" : "7.6.2", "build_flavor" : "default", "build_type" : "tar", "build_hash" : "ef48eb35cf30adf4db14086e8aabd07ef6xxx", "build_date" : "2020-03-26T06:34:37.794943Z", "build_snapshot" : false, "lucene_version" : "8.4.0", "minimum_wire_compatibility_version" : "6.8.0", "minimum_index_compatibility_version" : "6.0.0-beta1" }, "tagline" : "You Know, for Search" }
部署自建Kibana，并准备测试数据。
本文使用的自建Kibana版本为7.6.2，1个数据节点，具体操作步骤如下：
连接ECS服务器。
具体操作请参见使用Workbench登录Linux实例。
说明本文档以普通用户权限为例。
下载Kibana软件安装包并解压缩。
wget https://artifacts.elastic.co/downloads/kibana/kibana-7.6.2-linux-x86_64.tar.gz tar -zvxf kibana-7.6.2-linux-x86_64.tar.gz修改Kibana配置文件config/kibana.yml，增加
server.host: "0.0.0.0"配置，允许通过公网IP访问Kibana。进入Kibana安装目录，修改kibana.yml。
cd kibana-7.6.2-linux-x86_64 vi config/kibana.yml将
server.host的值修改为"0.0.0.0"以允许远程连接，修改后的配置文件关键内容为：# Kibana is served by a back end server. This setting specifies the port to use. #server.port: 5601 # Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values. # The default is 'localhost', which usually means remote machines will not be able to connect. # To allow connections from remote users, set this parameter to a non-loopback address. #server.host: "localhost" server.host: "0.0.0.0" # Enables you to specify a path to mount Kibana at if you are running behind a proxy. # Use the `server.rewriteBasePath` setting to tell Kibana if it should remove the basePath # from requests it receives, and to prevent a deprecation warning at startup. # This setting cannot end in a slash. #server.basePath: ""使用非root用户启用Kibana。
sudo nohup ./bin/kibana &登录Kibana控制台，添加示例数据。
通过公网IP地址登录Kibana控制台。
公网IP地址为：http://<ECS服务器的公网IP地址>:5601/app/kibana#/home。
在Kibana控制台首页，单击Try our sample data。
在 Sample data 页签下，找到 Sample web logs 卡片，单击其底部的 Add data 按钮，添加示例数据。
部署自建Logstash。
本文使用的Logstash版本为7.10.0，1个节点，具体操作步骤如下：
连接ECS服务器。
具体操作请参见使用Workbench登录Linux实例。
说明本文档以普通用户权限为例。
回到根目录，下载Logstash软件安装包并解压缩。
cd ~ wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.0-linux-x86_64.tar.gz tar -zvxf logstash-7.10.0-linux-x86_64.tar.gz修改Logstash的堆内存使用。
Logstash默认的堆内存为1 GB，您需要根据ECS规格配置合适的内存大小，加快集群数据的迁移效率。
进入Logstash的安装目录下，修改Logstash配置文件config/jvm.options，增加-Xms8g和-Xmx8g。
cd logstash-7.10.0 sudo vi config/jvm.options## JVM configuration # Xms represents the initial size of total heap space # Xmx represents the maximum size of total heap space -Xms8g -Xmx8g ################################################################ ## Expert settings ################################################################ ## ## All settings below this section are considered ## expert settings. Don't tamper with them unless ## you understand what you are doing ## ################################################################ ## GC configuration -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly ## Locale修改Logstash批量写入记录条数。
每批量写入5~15 MB数据，可以加快集群数据的迁移效率。
修改Logstash配置文件config/pipelines.yml，将每批量写入记录条数pipeline.batch.size从125改为5000。
vi config/pipelines.yml# # The path from where to read the configuration text # path.config: "/etc/conf.d/logstash/myconfig.cfg" # # # How many worker threads execute the Filters+Outputs stage of the pipeline # pipeline.workers: 1 (actually defaults to number of CPUs) # # # How many events to retrieve from inputs before sending to filters+workers pipeline.batch.size: 5000 # # # How long to wait in milliseconds while polling for the next event # # before dispatching an undersized batch to filters+outputs # pipeline.batch.delay: 50 # # # Internal queuing model, "memory" for legacy in-memory based queuing and # # "persisted" for disk-based acked queuing. Defaults is memory # queue.type: memory验证Logstash功能。
通过控制台输入输出收集数据。
bin/logstash -e 'input { stdin { } } output { stdout {} }'在控制台中输入"Hello world!"。
正常情况下，控制台会输出"Hello world!"。
[elastic@vm01 logstash-7.10.0]$ bin/logstash -e 'input { stdin { } } output { stdout {} }' Using bundled JDK: /home/elastic/logstash-7.10.0/jdk OpenJDK 64-Bit Server VM warning: Option UseConcMarkSweepGC was deprecated in version 9.0 a WARNING: An illegal reflective access operation has occurred WARNING: Illegal reflective access by org.jruby.ext.openssl.SecurityHelper (file:/tmp/jruby WARNING: Please consider reporting this to the maintainers of org.jruby.ext.openssl.Securit WARNING: Use --illegal-access-warn to enable warnings of further illegal reflective access WARNING: All illegal access operations will be denied in a future release Sending Logstash logs to /home/elastic/logstash-7.10.0/logs which is now configured via log [2022-03-21T15:39:24,470][INFO ][logstash.runner ] Starting Logstash {"logstash.ve inux-x86_64]"} [2022-03-21T15:39:24,606][INFO ][logstash.setting.writabledirectory] Creating directory {:s [2022-03-21T15:39:24,618][INFO ][logstash.setting.writabledirectory] Creating directory {:s [2022-03-21T15:39:24,845][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines [2022-03-21T15:39:24,865][INFO ][logstash.agent ] No persistent UUID file found. [2022-03-21T15:39:25,961][INFO ][org.reflections.Reflections] Reflections took 36 ms to sca [2022-03-21T15:39:26,356][INFO ][logstash.javapipeline ][main] Starting pipeline {:pipel s"}>["config string"], :thread=>"#<Thread:0x75693a9 run>"} [2022-03-21T15:39:26,997][INFO ][logstash.javapipeline ][main] Pipeline Java execution i [2022-03-21T15:39:27,032][INFO ][logstash.javapipeline ][main] Pipeline started {"pipeli The stdin plugin is now waiting for input: [2022-03-21T15:39:27,073][INFO ][logstash.agent ] Pipelines running {:count=>1, : [2022-03-21T15:39:27,211][INFO ][logstash.agent ] Successfully started Logstash A Hello world! { "host" => "vm01", "@version" => "1", "message" => "\"Hello world!\"", "@timestamp" => 2022-03-21T07:39:46.598Z }
（可选）步骤二：迁移索引元数据（设置和映射）
在进行数据迁移时，Logstash会帮助您自动创建索引，但是自动创建的索引可能与您待迁移的索引存在差异，导致迁移前后数据的格式不一致。因此建议您在数据迁移前，在阿里云Elasticsearch中手动创建目标索引，确保迁移前后索引数据完全一致。
您可以通过Python脚本创建目标索引，具体操作步骤如下：
连接ECS服务器。
具体操作请参见使用Workbench登录Linux实例。
说明本文档以普通用户权限为例。
创建并打开Python脚本文件（本文以indiceCreate.py为例）。
sudo vi indiceCreate.py修改Python脚本文件，拷贝以下代码（注意修改集群的连接地址、用户名和密码）。
#!/usr/bin/python # -*- coding: UTF-8 -*- # 文件名：indiceCreate.py import sys import base64 import time import httplib import json ## 源集群host。 oldClusterHost = "localhost:9200" ## 源集群用户名，可为空。 oldClusterUserName = "elastic" ## 源集群密码，可为空。 oldClusterPassword = "xxxxxx" ## 目标集群host，可在阿里云Elasticsearch实例的基本信息页面获取。 newClusterHost = "es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200" ## 目标集群用户名。 newClusterUser = "elastic" ## 目标集群密码。 newClusterPassword = "xxxxxx" 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 + " 或许是系统索引，不会重新创建，如有需要，请单独处理～")执行Python脚本，创建目标索引。
sudo /usr/bin/python indiceCreate.py参见登录Kibana控制台，登录目标集群的Kibana控制台，查看已创建的索引。
GET /_cat/indices?v
步骤三：迁移全量数据
连接ECS服务器。
具体操作请参见使用Workbench登录Linux实例。
在config目录下，创建并打开Logstash配置文件。
cd logstash-7.10.0/config vi es2es_all.conf参考以下配置，修改Logstash配置文件。
说明8.5版本Logstash的配置参数有所调整，本文同时列出了7.10.0版本和8.5.1版本Logstash的配置示例。
为了保证迁移数据的准确性，建议您创建多个Logstash管道配置文件，分批次迁移数据，每个Logstash迁移部分数据。
7.10.0版本
input{ elasticsearch{ # 源端ES地址。 hosts => ["http://localhost:9200"] # 安全集群配置登录用户名密码。 user => "xxxxxx" password => "xxxxxx" # 需要迁移的索引列表，多个索引以英文以逗号（,）分隔。 index => "kibana_sample_data_*" # 以下三项保持默认即可，包含线程数和迁移数据大小和Logstash JVM配置相关。 docinfo=>true slices => 5 size => 5000 } } filter { # 去掉一些Logstash自己加的字段。 mutate { remove_field => ["@timestamp", "@version"] } } output{ elasticsearch{ # 目标端ES地址，可在阿里云Elasticsearch实例的基本信息页面获取。 hosts => ["http://es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200"] # 安全集群配置登录用户名密码。 user => "elastic" password => "xxxxxx" # 目标端索引名称，以下配置表示索引与源端保持一致。 index => "%{[@metadata][_index]}" # 目标端索引type，以下配置表示索引类型与源端保持一致。 document_type => "%{[@metadata][_type]}" # 目标端数据的id，如果不需要保留原id，可以删除以下这行，删除后性能会更好。 document_id => "%{[@metadata][_id]}" ilm_enabled => false manage_template => false } }8.5.1版本
input{ elasticsearch{ # 源端ES地址。 hosts => ["http://es-cn-uqm3811160002***.elasticsearch.aliyuncs.com:9200"] # 安全集群配置登录用户名密码。 user => "elastic" password => "" # 需要迁移的索引列表，多个索引以英文以逗号（,）分隔。 index => "test_ecommerce" # 以下三项保持默认即可，包含线程数和迁移数据大小和Logstash JVM配置相关。 docinfo => true size => 10000 docinfo_target => "[@metadata]" } } filter { # 去掉一些Logstash自己加的字段。 mutate { remove_field => ["@timestamp","@version"] } } output{ elasticsearch{ # 目标端ES地址，可在阿里云Elasticsearch实例的基本信息页面获取。 hosts => ["http://es-cn-nwy38aixp0001****.elasticsearch.aliyuncs.com:9200"] # 安全集群配置登录用户名密码。 user => "elastic" password => "" # 目标端索引名称，以下配置表示索引与源端保持一致。 index => "%{[@metadata][_index]}" # 目标端数据的id，如果不需要保留原id，可以删除以下这行，删除后性能会更好。 document_id => "%{[@metadata][_id]}" ilm_enabled => false manage_template => false } }Elasticsearch input插件可以根据配置的查询语句，从Elasticsearch集群读取文档数据，适用于批量导入测试日志等操作。默认读取完数据后，同步动作会自动关闭，而阿里云Logstash需保证进程一直运行，关闭后将会重新启动进程，导致某些单一任务场景（如logstash input es）存在重复写数据的情况。设置长时间范围的定时任务可绕过写重复的情况，如每年3月5日13点20分触发任务执行，执行完第一次任务后停止管道运行，可避免重复写情况。可以通过cron语法配合schedule参数实现，详情请参见Logstash官网Scheduling介绍。
例如，设置3月5日13点20分执行任务：
schedule => "20 13 5 3 *"进入Logstash目录。
cd ~/logstash-7.10.0启动Logstash全量迁移任务。
nohup bin/logstash -f config/es2es_all.conf >/dev/null 2>&1 &
步骤四：迁移增量数据
连接ECS服务器，在config目录下，创建并打开Logstash增量配置文件。
cd config vi es2es_kibana_sample_data_logs.conf说明本文档以普通用户权限为例。
参考以下配置，修改Logstash配置文件。
7.10.0版本配置示例如下。
说明8.5版本Logstash的配置参数有所调整，需要去掉
document_type => "%{[@metadata][_type]}"。按如下脚本修改Logstash配置文件后，开启Logstash定时任务即可触发增量迁移。
input{ elasticsearch{ # 源端ES地址。 hosts => ["http://localhost:9200"] # 安全集群配置登录用户名密码。 user => "xxxxxx" password => "xxxxxx" # 需要迁移的索引列表，多个索引使用英文逗号（,）分隔。 index => "kibana_sample_data_logs" # 按时间范围查询增量数据，以下配置表示查询最近5分钟的数据。 query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}' # 定时任务，以下配置表示每分钟执行一次。 schedule => "* * * * *" scroll => "5m" docinfo=>true size => 5000 } } filter { # 去掉一些Logstash自己加的字段. mutate { remove_field => ["@timestamp", "@version"] } } output{ elasticsearch{ # 目标端ES地址，可在阿里云Elasticsearch实例的基本信息页面获取。 hosts => ["http://es-cn-zvp2m4bko0009****.elasticsearch.aliyuncs.com:9200"] # 安全集群配置登录用户名密码. user => "elastic" password => "xxxxxx" # 目标端索引名称，以下配置表示索引与源端保持一致。 index => "%{[@metadata][_index]}" # 目标端索引type，以下配置表示索引类型与源端保持一致。 document_type => "%{[@metadata][_type]}" # 目标端数据的id，如果不需要保留原id，可以删除以下这行，删除后性能会更好。 document_id => "%{[@metadata][_id]}" ilm_enabled => false manage_template => false } }重要Logstash记录的时间戳为UTC时间，如果您的本地时间为北京时间（东八区），那么两者会存在8个小时的时区差，此时将UTC时间转化为北京时间，可使用公式：UTC+时区差=北京时间。例如，以上示例中通过源端索引中的@timestamp字段进行range范围过滤查询获取增量数据，并在对应的时间上+8h转换为北京时间。
通过Logstash控制时间字段实现增量数据的同步，需确保原索引中有可控制的时间字段，如果原索引中没有时间字段数据，可使用ingest pipeline指定_ingest.timestamp获取元数据值，从而引入@timestamp时间字段。
进入Logstash目录。
cd ~/logstash-7.10.0启动Logstash增量迁移任务。
sudo nohup bin/logstash -f config/es2es_kibana_sample_data_logs.conf >/dev/null 2>&1 &在目标端Elasticsearch集群的Kibana中，查询最近更新的记录，验证增量数据是否同步。
以下示例的查询条件为：索引名称为kibana_sample_data_logs、最近时间范围为5分钟。
GET kibana_sample_data_logs/_search { "query": { "range": { "@timestamp": { "gte": "now-5m", "lte": "now/m" } } }, "sort": [ { "@timestamp": { "order": "desc" } } ] }
步骤五：查看数据迁移结果
查看是否完成全量迁移。
查看自建Elasticsearch集群的索引和数据量信息。
GET _cat/indices?v结果如下。
GET _cat/indices?v health status index uuid pri rep docs.count docs.deleted store.size pri.store.size green open .kibana_task_manager_1 CxAx5J2sT0qHPsWV 1 0 2 0 6.6kb 6.6kb green open .apm-agent-configuration dYz5bh4dTomjtDP3 1 0 0 0 283b 283b green open kibana_sample_data_logs PUBQrSkJRMGyI-cV 1 0 14074 0 11.6mb 11.6mb green open .kibana_1 MXhG2XbYTYSORB8G 1 0 49 4 139.5kb 139.5kb全量迁移前，查看阿里云Elasticsearch集群的索引和数据量信息。
GET _cat/indices?v health status index uuid pri rep docs.count docs.deleted store.size pri.store.size green open .aliyun-limiter-group 5K4N8YNUSxeJZCXPxxx 1 1 0 0 522b 261b green open .apm-agent-configuration vaVC28KVQMCsABwuxxx 1 1 0 0 522b 261b green open .monitoring-es-7-2022.03.19 9NUdZCaAQw-426Zrxxx 1 1 207485 15328 229.8kb 9.9kb green open highlight_unified PubNS7HIRR2B5FIfxxx 1 1 2 0 19.8kb 9.9kb green open .monitoring-es-7-2022.03.18 kEP-0LeeSh01-kg2xxx 1 1 117792 0 132.3mb 60.3mb green open .aliyun-limiter-config 6SJImN0bRoap3fYMxxx 1 1 0 0 522b 261b green open .kibana_1 0RRrLWLCT4aaT-1fxxx 1 1 27 4 20.8mb 10.4mb green open .security-7 D7Ux5eq7S5WtYH_Yxxx 1 1 55 0 397.7kb 198.4kb green open .monitoring-es-7-2022.03.21 n6DZS66KRmW1zaN7xxx 1 1 85969 5244 102.5mb 51.5mb green open .apm-custom-link SBnBUOojSd-Vt3xxxxx 1 1 0 0 522b 261b green open .monitoring-kibana-7-2022.03.20 eHPFB1h4Q8yxYbxAxxx 1 1 17278 0 5.8mb 2.8mb green open .kibana_task_manager_1 iDK1EK-iR22Gkhfxxxx 1 1 6 68 157kb 66.1kb green open .monitoring-kibana-7-2022.03.21 YIivw66dSBi0_Rwuxxx 1 1 6062 0 4.5mb 2.2mb green open kibana_sample_data_logs 1zaN5Ji7RWqbFwKZxxx 1 0 0 0 208b 208b green open .kibana-event-log-7.16.0-000001 ImZU-V4KRq2K3EUxxxx 1 1 1 0 11.4kb 5.7kb green open highlight_fvh sErtUXXpToiiPSaSxxx 1 1 2 0 23.5kb 11.7kb green open .monitoring-es-7-2022.03.20 SyOns3d-QU6ysbFDxxx 1 1 224781 43812 246.6mb 124.1mb green open .monitoring-kibana-7-2022.03.18 gOvcKvRlQ9O-PipQxxx 1 1 10700 0 3.4mb 1.7mb green open .monitoring-kibana-7-2022.03.19 IwSi_UIYQ5eFSyUJxxx 1 1 17280 0 5.8mb 2.9mb全量迁移后，查看阿里云Elasticsearch集群索引和数据量信息。
正常情况下，返回的记录条数应该与自建Elasticsearch集群一致。在 Kibana Dev Tools 中执行
GET _cat/indices?v命令，返回结果显示集群所有索引 health 均为 green、status 均为 open。其中kibana_sample_data_logs索引的 docs.count 为 14074，store.size 为 9.4mb，表明数据已成功迁移至目标集群。
查看是否完成增量迁移。
查看自建Elasticsearch集群的最近更新记录。
GET kibana_sample_data_logs/_search { "query": { "range": { "@timestamp": { "gte": "now-5m", "lte": "now/m" } } }, "sort": [ { "@timestamp": { "order": "desc" } } ] }返回结果如下。
{ "_source" : { "agent" : "Mozilla/5.0 (X11; Linux x86_64; rv:6.0a1) Gecko/20110421 Firefox/6.0a1", "bytes" : 658, "clientip" : "171.66.xxx", "extension" : "", "geo" : { "srcdest" : "CN:US", "src" : "CN", "dest" : "US", "coordinates" : { "lat" : 45.54039389, "lon" : -122.9498258 } }, "host" : "www.elastic.co", "index" : "kibana_sample_data_logs", "ip" : "171.66.xxx", "machine" : { "ram" : 3221225xxx, "os" : "win 7" }, "memory" : null, "message" : "171.66.xxx - - [2018-07-30T09:23:11.012Z] \"GET /security-analytics Gecko/20110421 Firefox/6.0a1\"", "phpmemory" : null, "referer" : "http://www.elastic-elastic-elastic.com/success/albert-sacco", "request" : "/security-analytics", "response" : 200, "tags" : [ "success", "security" ], "timestamp" : "2022-03-21T09:23:11.012Z", "url" : "https://www.elastic.co/solutions/security-analytics", "utc_time" : "2022-03-21T09:23:11.012Z", "event" : { "dataset" : "sample_web_logs" } }, "sort" : [ 1647854591012 ] }增量迁移完成后，使用同样命令查看阿里云Elasticsearch集群最近的更新记录。正常情况下，阿里云Elasticsearch集群的更新记录会与自建集群一致。