Elasticsearch Service 腾讯云 Elasticsearch 集群迁移之 Logstash 方案

一、迁移目标/范围

腾讯云 ES 是基于开源引擎打造的云端全托管 ELK（ES/Logstash/Kibana）服务，拥有高性能自研内核、支持读写分离、存算分离、自治索引、机器学习等能力，同时也支持自动弹性、零运维、低成本的 Serverless 版，被广泛应用于日志分析、在线搜索、向量检索与 RAG（检索增强生成）等场景。

本方案将详细介绍迁移流程、迁移步骤以及注意事项等方面的内容。通过本方案，用户可以在保证数据完整性、安全性和高效性的前提下，无需进行大规模业务改造，即可顺利平滑完成迁移过程。

二、方案简介

ES 迁移主推 Snapshot 方案，如有其他特殊场景，可参考 Logstash 及在线融合方案。具体方案对比如下：

三、概述

本迁移方案旨在为用户提供一种从自建 ES 或友商云 ES 迁移到腾讯云 ES 的实践教程。本方案将详细介绍使用 Logstash 快照方案的迁移流程、迁移步骤以及注意事项等方面的内容。通过本方案，用户可以在保证数据完整性、安全性和高效性的前提下，顺利完成迁移过程。

六、账号开通

客户需要开通一个腾讯云账号。

七、网络打通

客户的 ES 集群需要有一条能访问腾讯云服务的专线网络或 VPN。

·专线：专线接入

·VPN：建立 VPC 到 IDC 的连接（SPD 策略）

注意：数据迁移务必使用内网，否则默认会走公网进行数据传输，公网传输存在数据泄漏风险和潜在危险。

十、确认 ES 集群搭建前基础信息

十一、确认 ES 集群资源配置

生产环境的规格参考实际的 ES 业务需求，有特殊机型要求的，需提早申请，以便腾讯侧备货，腾讯云建议基础的 ES 生产环境配置如下：

十二、集群购买

购买腾讯云 ES 集群请参见 创建集群。

十三、Logstash 购买

购买腾讯云 Logstash 请参见 创建实例。

十四、插件安装及词典上传（必要）

在进行数据迁移前，需要确认索引是否使用自定义插件及自定义词典，如果有则需要在云上 ES 集群同步，否则相关索引数据恢复无法成功。

十五、插件查询

使用下面命令查询自建集群使用的插件：

curl -s -uelastic:'源端集群密码' 'http://源端IP:9200/_cat/plugins?h=component,version' | sort | uniq

十六、插件安装

腾讯云 ES 支持的插件列表及插件安装可以参考 腾讯云 Elasticsearch Service 插件列表，如果有另外的自定义插件需要安装可以联系腾讯云侧开通白名单使用此功能。

十七、词典查询

自定义词典查询如下：

curl -s -uelastic:'源端集群密码' 'http://源端IP:9200/_settings?pretty' | grep "_path"

自定义词典查询流程图

十八、IK 分词/停用词词典查询

# 进入到ES主目录，$ES_HOME需替换成实际目录
cd /$ES_HOME/config/analysis-ik
# 进入ik词典目录下，查看配置文件
cat IKAnalyzer.cfg.xml

未配置任何 ik 词典则默认内容如下，无需更新 ik 词典。如自定义了 ik 词典，则需上传至腾讯云 ES。

十九、词典上传

·同义词词典更新

进入 腾讯云 ES 控制台 > ES 集群管理 > ES 实例 > 高级配置界面，单击更新词典，上传自定义词典，配置索引属性即可使用。

·IK 分词/停用词词典更新

进入 腾讯云 ES 控制台 > ES 集群管理 > ES 实例 > 插件列表界面，选择 IK 插件单击更新词典，上传分词词典后即可。

21、方案介绍

·介绍：Logstash 是一款专门用于应用程序日志、事件的传输、处理、管理的产品。这里我们可以通过 Logstash 完成跨 ES 集群的数据迁移工作。

·适用场景：该种迁移方案适用于集群版本跨度较大，能够接受集群一段时间暂停写的场景。

22、迁移流程介绍

23、用户及角色迁移

·如需迁移用户及权限，可使用工具一键迁移：

rpm -vih https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/es-security-sync-lastest.x86_64.rpm

·或下载后上传再使用 rpm 安装：

rpm -vih es-security-sync-lastest.x86_64.rpm

注意：说明：

注意：security 功能 ES 6 版本才开始支持，ES 8 的 security 索引不允许修改。所以仅支持 6 到 7，以及 7 到 6 的迁移。

24、迁移示例

es-security-sync \\
–src_url http://10.0.xx.xx:9200 \\
–dst_url http://10.0.xx.xx:9400 \\
–src_user log_user \\
–dst_user log_user \\
–src_password ****** \\
–dst_password ******

25、索引模板迁移

·索引模板为 ES 业务使用中最常用的功能，在迁移前务必先同步索引模板。模板迁移使用工具一键迁移即可：

rpm -vih https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/es-template-cp-lastest.x86_64.rpm

·或下载后上传再使用 rpm 安装：

rpm -vih es-template-cp-lastest.x86_64.rpm

26、迁移示例

es-template-cp \\
–src_url http://源ES IP:9200 \\
–dst_url http://腾讯云ES IP:9200 \\
–src_password '******' \\
–dst_password '腾讯云ES密码'

27、集群管道迁移（按需操作）

ES 管道迁移使用工具一键迁移即可。

#下载jar包
#运行环境: JDK-1.8
wget https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/CopyPipelines.jar

28、示范样例

注意：任何参数不要空缺，无密码集群可指定任意密码。

·样例一

源端，目标端，集群带密码，跳过目标端已存在的同名管道。

java -jar CopyPipelines.jar http://源ES IP:9200 elastic '源端ES密码' http://腾讯云ES IP:9200 elastic '腾讯云ES密码' skip

·样例二

源端，目标端，集群带密码，覆盖目标端已存在的同名管道。

java -jar CopyPipelines.jar http://源ES IP:9200 elastic '源端ES密码' http://腾讯云ES IP:9200 elastic '腾讯云ES密码' overwrite

29、索引属性同步

·由于 logstash 不支持迁移 es 索引属性，例如 settings、mapping，所以需要提前同步：

rpm -vih https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/es-index-properties-sync-lastest.x86_64.rpm

·或下载后上传再使用 rpm 安装：

rpm -vih es-index-properties-sync-lastest.x86_64.rpm

注意：说明：

注意：该工具支持修改索引 setting、mapping。

30、同步示例

es-index-properties-sync \\
–src_url http://10.0.xx.xx:9200 \\
–dst_url http://10.0.xx.xx:9200 \\
–src_password ****** \\
–dst_password ****** \\
–indices '*,-.*' \\

注意：需要同步的索引在目标集群需不存在，否则会自动跳过已存在的索引，建议同步索引属性前清理目标集群索引。

32、索引分组

在索引数量比较大的场景下，我们需要将索引分成多个组，以启动多个实例或管道来加速迁移。我们提供了分组工具：

33、工具安装

rpm -vih https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/es-index-grouping-lastest.x86_64.rpm

34、参数说明

–user default: elastic
–password default: 默认读取环境变量 export password='***'
–ip default: 没有默认值，必要参数，不允许为空
–port default: 9200
–pattern default: 索引匹配规则，默认为 '*'，支持逗号分隔
–instances default: 分组数量。没有默认值，必要参数，不允许为空

35、使用示例

·示例 1

ES 实例没有密码认证，端口为9200，且所有索引都需要进行 分组，分组数为10个。

es-index-grouping –ip 源ES IP –instances 10

·示例 2

密码赋值和执行命令写成一条，中间无需分号。

password='123456' es-index-grouping –ip 源ES IP –port 9200 –user logUser –pattern '*,-.*' –instances 10

将分组后的 index list 根据不同管道填入 index =>，比如我们将全部索引分了10个组，我们则需要创建10个实例或管道。

36、Logstash 管道配置

1. 进入 Logstash 控制台，点击管道管理>新建管道。

2. 管道配置：单击引用模板可以快捷配置管道，配置完成后，单击保存并部署。

·pipeline.workers：一般为 CPU 数。

·pipeline.batch.size：设置批量执行 event 的最大值，一般3000-5000，不建议超过10000。

·pipeline.batch.delay：批处理的最大等待时长，一般3000-5000ms即可。

37、Config 配置

input {
elasticsearch {
hosts => "源ES IP:9200"
index => "*,-.*" # 如果是分组同步，这里则填入其中一个组的 index list，支持通配，例如：index => "index-a-*, index-b-*"
user => "elastic"
password => "xxxxxx"
docinfo => true
size => 5000
slices => 4 #建议设置为源端多数索引的分片数量
}
}
output {
elasticsearch {
hosts => ["http://腾讯云 ES IP:9200"]
user => "elastic"
password => "腾讯云ES密码"
index => "%{[@metadata][_index]}"
document_type => "%{[@metadata][_type]}"
document_id => "%{[@metadata][_id]}"
}
}

如果有索引改名的需求，可以使用 filter 插件进行一下关键词替换：

filter {
mutate {
gsub => ["[@metadata][_index]", "test-", "ceshi-"]
}
}

注意：说明：

注意：这里演示的是将 test- 替换为 ceshi-。

38、Logstash 增量同步

·logstash 本身不具备同步增量的能力，其增量的实现需要业务使用 query 语句完成，且需要业务数据有 update_time 之类的字段，以标识数据的更新时间。

·如果业务数据只有 create_time 或 @timestamp 这样的字段，则只能做到增量同步，无法做到差异同步（update）。

·logstash 不支持同步删除操作，所以在进行 logstash 增量同步后，目标端可能比源端多出部分数据（即logstash同步过程中源端被删除的数据）。

40、开启索引动态 mapping

#为单个索引开启动态mapping
PUT index_name/_mapping
{
"dynamic": true
}
#为全部索引开启动态mapping
PUT *,-.*/_mapping
{
"dynamic": true
}

41、业务改造

业务新增、更新数据时需要新增 update_time 字段，标识数据的当前更新时间。

42、增量同步

在按照这篇完成了 ES 全量数据同步后，继续配置新管道进行增量数据同步。

input {
elasticsearch {
# 源端ES地址。
hosts => ["http://源ES IP:9200"]
# 安全集群配置登录用户名密码。
user => "elastic"
password => "源端ES密码"
# 需要迁移的索引列表，多个索引使用英文逗号（,）分隔，支持通配。
index => "sample_data"
# 按时间范围查询差异数据，以下配置表示查询最近1小时更新的数据。
query => '{"query":{"range":{"update_time":{"gte":"now-1h/h","lte":"now/m"}}}}'
# 定时任务，以下配置表示每半小时执行一次。
schedule => "0 */30 * * * ?"
scroll => "5m"
docinfo=>true
size => 5000
}
output {
elasticsearch {
hosts => ["http://腾讯云 ES IP:9200"]
user => "elastic"
password => "腾讯云ES密码"
index => "%{[@metadata][_index]}"
document_type => "%{[@metadata][_type]}"
document_id => "%{[@metadata][_id]}"
}
}

43、特别注意

logstash 增量同步前建议充分调试，将 input 插件里的 query 语句拿到源端执行，看能否查到数据，例如：

GET sample_data/_search
{
"query": {
"range": {
"update_time": {
"gte": "now-1h/h",
"lte": "now/m"
}
}
},
"_source": "update_time"
}

以下两种情况可能导致查询不到相关数据：

·源端数据有延迟或者不是流式的，可能是跑批的数据，这样我们则可以把 now-1h 调大一些，比如 now-5h，甚至更大。我们也可以直接获取当前数据的最大时间来判断数据实时性，例如：

GET sample_data/_search
{
"sort": [
{
"update_time": {
"order": "desc"
}
}
],
"_source": "update_time"
}

·源端数据时间字段非 date 类型，而是 keyword 或者 text 类型，那么即便数据是时间字符串，也无法使用时间运算，更无法使用 range 查询，所以结果会为空。

·查到的数据比当前时间少8小时，则是符合预期的，因为 ES 默认是使用的 UTC 时间，无需对时间进行额外处理；

·如果查到的数据是当前的时间，没有少8小时，则说明数据在写入时候没有使用 UTC 时间，则同步时需要在 logstash 侧使用 now 运算：now + 8h。如果数据本身有延迟，再在8h上额外增加即可。

·如果查询报错，则说明使用的时间字段不是时间类型，而是数值类型或其他类型，例如：

44、数据校验

可以下载数据条数校验工具进行对比，工具仅对比两端集群索引文档数量，不会校验内容是否相同。

45、工具安装

#下载jar包
#运行环境: JDK-1.8
wget https://tools-release-1253240642.cos.ap-shanghai.myqcloud.com/elasticsearch/packages/ElasticsearchCompare.jar

47、示范样例

共需要如下输入7个参数，任何参数不能空缺，无密码集群可指定任意密码。

java -jar ESIndexDocumentCountComparison.jar <srcClusterUrl> <srcUser> <srcPassword> <dstClusterUrl> <dstUser> <dstPassword> <index pattern>

样例

源端，目标端，集群带密码，需要校验两端集群全部索引的数据一致性，如果集群无密码则可输入任意密码。

java -jar ESIndexDocumentCountComparison.jar http://源端ES IP:9200 elastic '源端ES密码' http://腾讯云ES IP:9200 elastic '腾讯云ES密码' "*,-.*"

48、返回说明

工具首先会返回源端集群的 close 索引，close 状态索引无法检验数据一致性，然后返回对于两边集群数据不一致的索引信息，不返回数据一致的索引信息。

·数据不一致返回

·数据一致返回

49、业务切割

数据迁移完成并完成一致性校验，以及完成业务兼容性完全测试，则可以进行业务切割上云，将 es ip 切换为云上内网访问地址即可：