Elasticsearch 数据迁移与容灾实践，看这一篇就够了！

admin

102359
文章

87
评论

2021年7月25日03:37:47评论280 views字数 5808阅读19分21秒阅读模式

说明：本文为高斌龙老师在 GOPS 2021 · 深圳站的现场分享整理而成。

作者简介
高斌龙（bellen），腾讯云大数据开发工程师，目前专注于 ElasticSearch 云产品研发工作

今天跟大家分享的主题是“ElasticSearch数据迁移与容灾实践”。

内容的来源是过去几年我们在服务腾讯云客户使用 ES 过程中碰到的数据同步、数据迁移以及一些容灾方面的需求，我们同样也给出了解决办法，所以今天的分享内容分为三个部分，首先介绍一下 ES。

ES 是开源的基于 Lucene 打造的分布式搜索引擎，根据 DB-engines 数据库排名，ES常年稳定在七八位，ES应用场景典型的有搜索、日志，在 APM、IOT 领域也有 ES 身影。

一、异构数据与 ES 同步

第一部分异构数据与ES同步，指不同数据源与 ES 之间数据导入导出。

不同的数据源，像关系型数据库 MySQL、PostgreSQL，或者文档数据库 MongoDB，消息队列 Kafka、RabbitMQ 和 ES 间的同步，还有 Hadoop 生态系统与 ES 的数据相互的导入导出。

也有客户想把 ES 本身的数据归档到对象存储、腾讯云 COS、阿里云OSS，或亚马逊的 S3 的对象存储中，从而降低 ES 本身的存储成本。因为对象存储可以称为廉价存储，存储成本比较低。

异构数据同步有个非常典型的场景就是电商搜索场景，电商数据本身是要存储在关系型数据库像比较重要的 MySQL 或 Oracle 中，这些数据是作为主库的，为了加快商品条目的搜索，就需要用到 ES，需要把 MySQL 中的数据实时同步到 ES 中，利用 ES 的搜索能力进行商品搜索。

MySQL 数据要导入到 ES ，根据同步的方式可以分为两种。

第一种是离线同步，是指源库 MySQL 的数据暂停写入，就是说数据不再新增，可以称为一次性或全量同步，这个时候就可以采用 Logstash 组件，在 input 端通过 JDBC Driver 去连接 MySQL 数据库批量地把数据读出来，在 output 端利用 API 进行批量写入，但是特点是一次性的，只能进行全量导入。

在某些场景下，如果源库里的数据没有删除或者数据量本身比较小的情况下我们也可以做简单增量，即源库数据没有删除，数据量比较小只有新增或更新，这个时候可以根据数据库里的 updateTime 字段批量定时拉取最近一段时间内的新增数据，把新增数据也导入到 ES 中，但是问题是同步的实时性不是特别好，速率比较低。

如果要真正做到实时同步就要基于 MySQL 的 Binlog，当然也有一些开源组件如 canal/Mypipe，用这些组件实时获取到日志进行解析，也可以直接吐出到 ES 中，也可以加一层消息队列把解析到的 Binlog 中的操作数据做持久化，可以保证实时同步数据的可靠性，数据在 Kafka 中最常见的是采用 Logstash 进行消费写入到 ES。

与MySQL类似，如果 MongoDB 文档数据库写入到ES，也可以采用 Logstash 一次性全量把 MongoDB 中的数据写入到 ES。

如果要进行实时同步的话，就要采用 MongoDB 的 OPlog，开源的 Monstache 进行实时同步 MongoDB 数据，Monstache 也可以支持全量同步，如果源库中的数据跑了很长一段时间后数据积累了很多，这个时候 Monstache 先进行一次全量再进行增量再把数据写入到 ES 中去。

第三种常见的是在ELK日志系统中，通常我们采集日志不是直接把日志采集就直接放到ES中，而是要做持久化，将日志经过 Logstash 等采集到之后，先吐到消息队列 Kafka 中去，再采用 Logstash 消费 Kafka 中的数据写入ES。

除了用 Logstash 消费 Kafka 数据，腾讯云上还有SCF无服务器云函数，也可以实现对 Kafka 数据的消费和写入ES。

这个云函数是跑在容器里的，可以订阅到 Kafka 中的某一个 topic，当这个 topic 产生数据之后之后才可以触发 SCF 云函数的执行，所以它的成本是非常低的，因为是 Serverless，根本不会用到服务器资源，可能跑了几百万次费用才几块钱，成本是非常低的。

第四种是对象存储，有一些客户需要把对象存储中的数据同步到 ES 中，对象存储本身有bucket存储桶，里面可以存放各种内容，包括普通文件和音视频文件等，如果要进行全量同步的话，可以采用 Logstash。

如果增量同步，目前腾讯云上也是采用SCF云函数，可以订阅到对象存储里某一个bucket，如果这个bucket上传文件的话，就可以触发到云函数的执行，云函数内部逻辑会把文件读出来做一些解析然后再写入到ES中。

第五种是 Hadoop 生态系统和 ES 数据的导入导出，这种场景 ES 官方提供了 ES-Hadoop 组件，可以实现 Hadoop 各种生态内的组件和ES的交互，可以在 hive中建一个外部表，指向的是ES本身的索引，就可以使用 hivesql 查询其中的数据，也可以通过这个组件到把数据导出到HDFS中。

还有一种场景是ES本身的数据磁盘上的索引文件可以通过快照的方式可以归档中 HDFS 中，因为一些老的数据做归档存储在ES中的索引文件都可以被删掉了，真正需要用时再恢复到ES中，从而可以降低ES本身的存储成本。

二、ES 集群间数据迁移

介绍完ES和异构数据的数据同步后，接下来介绍一下ES本身不同集群之间的数据迁移，这部分内容主要是为了满足客户跨机房数据同步或跨云数据同步的需求。

根据迁移方式的不同分为离线迁移和在线迁移。离线迁移是迁移过程中旧的集群可以停服或者暂停写入，因为增量数据不好进行同步。如果客户可以接受离线迁移的话需要把原集群暂停写入，迁移完成后业务切换新的集群进行读写。

如果客户不接受就得进行在线迁移，迁移过程中旧的集群不能停服不能暂停写入，这部分增量的数据也需要进行迁移。

离线迁移工具非常多，Elasticsearch-dump，是用 Node.js写的，特点是简单易用，适合数据量比较小的场景，10GB 以下的数据量建议使用该工具，因为它本身的稳定性不是特别好。

第二种是 Logstash，比较适合于对数据进行过滤或预处理的场景，或者源集群与目的集群版本跨度较大的场景，像刚才赵班长（舜东）提的集群是1.4 的版本，这个时候就可以经过 Logstash 从 1.4 的版本迁移到高版本，比如 5.x，6.x的版本。

第三种是 Reindex，是 ES 本身提供的 API，通过直接调用ES的API方式进行迁移。不仅可以做同集群之间不同索引的迁移，也可以做跨集群之间的数据迁移。但是有一个限制，目的端的集群需要能够访问源端集群的节点。Logstash、Reindex 适合于 100GB 以下的数据量。

2.1 离线迁移

如果数据量比较大，到了几十或几百TB，用前面的方式就比较慢了，通过批量读+批量写的方式效率是非常低的，这时候就可以采用 Snapshot，是ES本身提供的一个API，可以直接把源集群节点磁盘上的所有文件进行备份，可以备份到 COS 或者 HDFS 中去。

然后在目的端集群进行恢复，恢复的过程比较快，所以适合数据量比较大对迁移速度要求比较高的场景。

2.2 在线迁移

接下来介绍在线迁移。根据用户写入场景可以分为两种，第一种是用户的写入只有数据追加或更新没有删除，在数据量比较小的情况下就可以使用 Logstash 先进行全量再进行增量迁移，全量是指一次性把某一个索引从源集群导入到目的端集群，然后再通过 Logstash 的配置文件里增加一个定时任务，目的是比如每隔一分钟拉取过去一分钟内源集群内新增的数据。

拿到新增数据再把这部分增量数据写入到目的端 ES 中，它迁移速率比较低，只适合数据量比较小的场景。如果有删除操作的话，Logstash 根本就获取不到删除操作，就需要用其他方案。

第一种方案是 snapshot + 双写。snapshot 是对 ES 底层文件进行备份，所以可以通过snapshot把一些存量的、老的索引进行一次性迁移，它的迁移速度还是比较快的；增量的索引可以采用双写，同时向两个索引进行写入，保证双写的数据在两个集群里都是一致的。

第二种是业务端进行双写。如果业务端数据都存储在消息队列 kafka 中的话，这时可以起两组 Logstash 同时消费 Kafka 中的数据，然后写入两个 ES 中，这样也可以保证数据的一致性。

第三种是采用 ES 原生的 CCR 跨集群复制功能，开启集群间的数据同步。优点是数据一致性可以得到很好保证。

第四种是腾讯云提出的采用节点双网卡的方案，可以使源集群和目的集群融合，使得这两个集群成为一个集群。再采用 ES 的 exclude node 功能把老的集群中数据搬迁到新集群中，之后再把老节点下掉，就完成了一次迁移。

迁移用的双写的话，一般业务数据可以存储在消息队列 Kafka 中，起一组 Logstash 并行消费来写入到两个 ES 中，优点是写入过程中丢失数数据的风险比较低，可以保证新旧数据的一致性。

如果采用 CCR 的话有一定限制，因为 CCR 是在高版本集群中才会支持，要求新集群和旧集群的版本在6.5以上，并且都包含高级特性。

比如 ClusterA 作为一个 Leader，ClusterB 是一个 Follower，ClusterA 里的一些 index是 Leader index，B 中的索引是 Follower Index，Follower Index 会定期向 Leader Index 主动拉最近一段时间内数据的更新，然后在自己这一端做重放，实际上类似于MySQL 的 binlog。

但是 Leader index 必须开启 soft deletes 配置。6.x以上的版本是默认不开启的，如果是已经创建好的索引则不能采用CCR，7.x以上的版本默认这个特性是开启的，所以可以直接采用 CCR。