Elasticsearch 2.3.4 低版本数据迁移方案

方案	适用场景	结果
COS 快照	数据量大的场景（GB、TB、PB 级别）对迁移速度要求较高的场景	失败
Logstash	迁移全量或增量数据，且对实时性要求不高的场景需要对迁移的数据通过 es query 进行简单的过滤的场景需要对迁移的数据进行复杂的过滤或处理的场景版本跨度较大的数据迁移场景，如 5.x 版本迁移到 6.x 版本或 7.x 版本	失败
Elasticsearch-dump	数据量较小的场景	失败
Elasticsearch Reindex API	不确定是否因为老版本不支持远程remote配置	失败
Linux Rsync	直接将文件同步过去，感觉可以做到，但目前数据源不能停机，故无法使用	失败
自实现	自己写代码，调用 es http api 比较自由，只要不断线，都好说，可能数据量大就不好搞，不过做增量的话够用了	成功

先说结果，以上四种方案，都不适用，目前是es不能停机，然后需要增量迁移，版本也比较古老，不大好操作

Elasticsearch-dump

比较简单，安装好 node.js，再安装 elasticsearch-dump 插件，即可使用，可以端到端传输，缺点也是速度慢，而且容易断开连接，不推荐
参考：https://www.shuzhiduo.com/A/kmzL2kNl5G/

# 拷贝索引
elasticdump--input=http://production.es.com:9200/my_index--output=http://staging.es.com:9200/my_index--type=mapping# 拷贝数据
elasticdump--input=http://production.es.com:9200/my_index--output=http://staging.es.com:9200/my_index--type=data	# 拷贝所有索引
elasticdump--input=http://production.es.com:9200/--output=http://staging.es.com:9200/--all=true

Logstash

比较简单，安装好 Logstash，配置一下源地址和目标地址，即可使用，但缺点也是速度慢，然后也容易部分数据报错，但比Elasticsearch-dump 好使一点，比如14.6w笔，可能成功个14w笔
参考：https://zhuanlan.zhihu.com/p/479468405

vi logstash.confinput {elasticsearch {hosts => ["http://10.92.60.13:9201"]  # 源 Elasticsearch 集群的主机地址和端口index => "schedulezzb"       # 源索引的名称docinfo => truequery => '{ "query": { "match_all": {} } }'  # 可选：定义要迁移的数据查询scroll => "10m"                # 可选：定义滚动查询的保持时间size => 500                  # 可选：每次滚动查询返回的文档数量ssl => falsecodec => { charset => "UTF-8" }}
}## 过滤 type，仅 yuqingSubmit
filter {if [@metadata][_type] == "yuqingSubmit" {mutate {remove_field => ["[@metadata][_index]", "[@metadata][_type]", "[@metadata][_id]"]}}
}output {elasticsearch {hosts => ["http://10.92.60.113:9201"]  # 目标 Elasticsearch 集群的主机地址和端口index => "schedulezzb_new"       # 目标索引的名称document_id => "%{[@metadata][_id]}"  # 可选：保留源文档的 ID，这样就实现了增量更新}
}

启动

logstash-2.3.3/bin/logstash -f logstash.conf

Reindex API

源数据：10.92.60.13
目标数据：10.92.60.113

提示该版本不支持 remote 方式，另外需要注意的是 2 个索引名字不能一致，可以导完之后再通过设置别名访问

POST http://10.92.60.113:9201/_reindex
{"source": {"remote": {"host": "http://10.92.60.13:9201"},"index": "schedulezzb"},"dest": {"index": "schedulezzb_new"}
}

Linux Rsync

参考：https://www.ruanyifeng.com/blog/2020/08/rsync.html

示例：将etc推送到10.10.10.10下的自定义文件夹下
rsync -az /etc/ 10.10.10.10:/opt/etc-$(hostname)-$(date +%F)## 【不生效】ES文件到目标，3台，其中，12上面没数据所以不迁移，112又有(很奇怪)，配置显示13和12都是master节点
13：/data/10.92.60.16/data/elk/elasticsearch-2.3.4-ga/data/elasticsearch/nodes/0/indices/schedulezzb
12：/data/10.92.60.15/data/elk/elasticsearch-2.3.4-ga/data/elasticsearch/nodes/0/indices/schedulezzb
11：/data/10.92.60.14/data/elk/elasticsearch-2.3.4-ga/data/elasticsearch/nodes/0/indices/schedulezzb111，112，113：/home/elasticsearch-2.3.4-ga/data/elasticsearch/nodes/0/indices/schedulezzb
## 13->113，11->111，操作前备份：cp -r elasticsearch-2.3.4-ga/ elasticsearch-2.3.4-ga-bak20230408
删除113，111的data目录
rsync -az /data/10.92.60.16/data/elk/elasticsearch-2.3.4-ga/data 10.92.60.113:/home/elasticsearch-2.3.4-ga
rsync -az /data/10.92.60.14/data/elk/elasticsearch-2.3.4-ga/data 10.92.60.111:/home/elasticsearch-2.3.4-ga做完后重启后数据还是不生效，估计是由于数据源没有停机导致复制过去的文件不连续导致的

代码自实现

有2种方式，一种是通过es提供的jar包实现连接，TransportClient，通过连接9300实现。还有一种是通过Http连接9200实现，总的来说，9300比较快，但比较容易断开连接，最终使用的是9200，稳定一点
具体代码：https://gitee.com/zhaomingjian/workspace_luoan_demo/tree/master/spring-boot-elasticsearch
不过使用 http 的话，需要注意方式，刚开始采用分页的方式，遇到了 es 分页深度不能超过10000 的限制，后续改成了查询 ids 的方式，每次查询 50 个 id，最终愉快的实现了需求

package com.example.springbootelasticsearch.http;import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;import java.util.ArrayList;
import java.util.List;/*** 基于各种方案都不能用，尝试使用 http** @author jason*/
@Slf4j
public class HttpClientIds {private static final String INPUT_QUERY_TEMPLATE_IDS_JSON = "{\"size\":50,\"_source\":[\"id\",\"title\",\"url\",\"siteName\",\"mediaId\",\"publishTime\",\"content\",\"createUser\",\"reportState\",\"state\",\"createUserGroupId\",\"discoveryUnit\",\"discoveryPerson\"],\"query\":{\"bool\":{\"filter\":[{\"ids\":{\"values\":{}}}]}}}\n";private static final String INPUT_JSON = "D:/schedulezzb-yuqingSubmit-input-json.txt";private static final String INPUT_LOG = "D:/schedulezzb-yuqingSubmit-input.log";private static final String OUTPUT_LOG = "D:/schedulezzb-yuqingSubmit-output.log";/*** 数据迁移* 数据源：http://10.92.60.13:9201/schedulezzb/yuqingSubmit/_count*/public static void main(String[] args) {
//        inputListStart();outputListStart();}private static void outputListStart() {FileUtil.writeUtf8String("", OUTPUT_LOG);String schedulezzbYuqingSubmit = FileUtil.readUtf8String(INPUT_JSON);List<String> stringList = JSONUtil.toList(schedulezzbYuqingSubmit, String.class);for (String str : stringList) {JSONObject _source = JSONUtil.parseObj(str);output113(_source);}}private static JSONArray inputListStart() {FileUtil.writeUtf8String("", INPUT_LOG);int oldCount = 145433;int newCount = 141027;int cutCount = oldCount - newCount + 1;List<String> idAllList = new ArrayList<>(cutCount);for (int i = newCount; i <= oldCount; i++) {idAllList.add(i + "");}List<List<String>> partition = ListUtil.partition(idAllList, 50);System.out.println("总页数：" + partition.size());JSONArray resultAllArray = new JSONArray();for (List<String> subIds : partition) {// 根据 pageNum 查询得到的结果集JSONArray resultArray = inputList(subIds);resultAllArray.addAll(resultArray);System.out.println("当前进度：" + resultAllArray.size());}FileUtil.writeUtf8String(resultAllArray.toString(), INPUT_JSON);System.out.println("实际差值：" + cutCount);System.out.println("执行是否成功：" + (cutCount == resultAllArray.size()));return resultAllArray;}private static JSONArray inputList(List<String> subIds) {String url = "http://10.92.60.13:9201/schedulezzb/yuqingSubmit/_search";String request = StrUtil.format(INPUT_QUERY_TEMPLATE_IDS_JSON, subIds);// HTTPString response = HttpUtil.post(url, request);// 数据处理JSONObject responseJson = JSONUtil.parseObj(response);JSONObject hits = responseJson.getJSONObject("hits");JSONArray hits2 = hits.getJSONArray("hits");if (hits2.size() == 50 || hits2.size() == 7) {} else {FileUtil.appendUtf8String("入参：" + request, INPUT_LOG);FileUtil.appendUtf8String("出参：" + hits2, INPUT_LOG);FileUtil.appendUtf8String("\n数量有误：" + hits2.size(), INPUT_LOG);}JSONArray resultArray = new JSONArray();for (int i = 0; i < hits2.size(); i++) {JSONObject jsonObject = hits2.getJSONObject(i);JSONObject _source = jsonObject.getJSONObject("_source");resultArray.add(_source);}return resultArray;}/*** 13数据写入113*/private static void output113(JSONObject _source) {String id = _source.getStr("id");String url = "http://10.92.60.113:9201/schedulezzb/yuqingSubmit/" + id;String request = _source.toString();// HTTPString response = HttpUtil.post(url, request);System.out.println("数据写入-出参：" + response);FileUtil.appendUtf8String("数据写入-入参：" + request, OUTPUT_LOG);FileUtil.appendUtf8String("数据写入-出参：" + response + request, OUTPUT_LOG);}}

ps：重启es步骤

kill -9 进程id
su es
./elasticsearch -d