Elasticsearch 数据离线迁移方案（含冷热分层场景）

原创

岳涛

修改于 2023-06-25 14:40:56

2.9K0

修改于 2023-06-25 14:40:56

文章被收录于专栏：大数据生态

说明

本文描述问题及解决方法同样适用于 腾讯云 Elasticsearch Service（ES）。

迁移方案

logstash（需要打通专线，适合通用场景，无版本要求，支持跨大版本迁移）
snapshot（网络无需互通，首推方案，支持增量，有版本要求）

前置工作

在正式迁移数据之前，我们需要手动同步源端与目标端的 template，以避免索引 mapping 不符合预期。通常情况下 settings 是需要调整的，比如节点数发生改变，或者需要增加一些其他配置。

1、Logstash迁移（含冷热分层场景）

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

适用场景：该种迁移方案适用于集群规模比较大，能够接受集群一段时间暂停写的场景。

logstash插件说明：

Logstash 是插件式工作模式，他的插件主要分为 3 种 input/filter/output

INPUT PLUGIN # 把数据搜集进来的插件

FILTER PLUGIN # 过滤掉我们不需要的数据

OUTPUT PLUGIN # 将数据输出至ES，当然可以保存本地或者redis

迁移示例

1. 使用elasticsearch-dump同步索引属性

elasticsearch-dump 是一款开源工具，github 地址

由于 logstash 不支持迁移 es 索引的一些属性，比如 settings、mapping，所以需要借助第三方工具来进行同步索引属性。

安装方式：

npm install elasticdump -g

同步示例：

# 同步settings
elasticdump --input='http://username:passwd@1.1.1.1:9200/{index_name}' --output='http://username:passwd@2.2.2.2:9200/{index_name}' --type=settings
# 同步mapping
elasticdump --input='http://username:passwd@1.1.1.1:9200/{index_name}' --output='http://username:passwd@2.2.2.2:9200/{index_name}' --type=mapping

注意：

密码如果包含特殊符号，则需要进行 URL 编码转换
如果索引比较多，可以使用shell遍历提交

for index_name in `cat index_list.txt`
do
  elasticdump --input="http://username:passwd@1.1.1.1:9200/${index_name}" --output="http://username:passwd@2.2.2.2:9200/${index_name}" --type=settings
  elasticdump --input="http://username:passwd@1.1.1.1:9200/${index_name}" --output="http://username:passwd@2.2.2.2:9200/${index_name}" --type=mapping
done

2. 将目标端创建好的空索引预先进行冷热分层（如无冷热分层可忽略这步）

1）elasticsearch版本 >= 7.10

指定热数据：

PUT *logstash-*/_settings
{
  "routing": {
    "allocation": {
      "include": {
        "_tier_preference": "data_hot,data_warm,data_cold"
      }
    }
  }
}

指定冷数据：

PUT *logstash-*/_settings
{
  "routing": {
    "allocation": {
      "include": {
        "_tier_preference": "data_warm,data_hot,data_cold"
      }
    }
  }
}

2） elasticsearch版本 < 7.10

指定热数据：

PUT *logstash-*/_settings
{
  "routing": {
    "allocation": {
      "include": {
        "require.temperature": "hot"
      }
    }
  }
}

指定冷数据：

PUT *logstash-*/_settings
{
  "routing": {
    "allocation": {
      "include": {
        "require.temperature": "warm"
      }
    }
  }
}

3. 配置logstash管道

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "*"
        docinfo => true
        size => 5000
        scroll => "5m"
      }
}

output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        user => "elastic"
        password => "your_password"
        index => "%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
    }
}

上面的配置文件将源端 ES 集群的所有索引迁移到目标 ES 集群中。当然，我们也可以只迁移某些特定的索引。

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

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

这里演示的是将 test-替换为 ceshi-

filter 的详细使用说明，可以参考一下filter插件官方文档。

4. 优化logstash运行参数

1）pipeline.workers

该参数可控制 output 或 filter 插件的工作线程数（只能设置正整数），当发现事件正在备份或 CPU 没有饱和，则可以增加工作线程，以提高性能。一般为 CPU 线程数。

2）pipeline.batch.size

设置批量执行 event 的最大值，该值是用于 input 的批量处理事件值，再打包发送给 filter 和 output，增加该值可以在一定范围内提高性能，但是需要增加额外的内存开销。

3）pipeline.batch.delay

批处理的最大等待值（input 需要按照 batch 处理的最大值发送到消息队列，但是不能一直等，所以需要一个最大的超时机制）。

优势：

1. 使用场景最为通用，既可以迁移少量数据，也可以迁移较大规模的数据量；

2. 支持在迁移过程中做一些筛选、拆分、合并；

3. 支持高版本往低版本迁移。

不足：

1. Logstash 的数据迁移前，需要提前手动设置好索引的 settings 和 mapping 信息，不支持一键迁移；

2. Logstash 在迁移过程中节点宕机或进程异常退出，再次启动时，会从头开始读取数据，可能会出现重复数据的情况；

3. logstash 不支持同步更新/删除的动作；

4. 相对快照，迁移速度较慢。

扩展文档：Logstash官方文档；

注意：

logstash工具本身不具备自动同步增量的能力，其增量的实现需要业务使用query语句完成，增量同步demo：

        # 源端ES地址。
        hosts =>  ["http://localhost:9200"]
        # 安全集群配置登录用户名密码。
        user => "xxxxxx"
        password => "xxxxxx"
        # 需要迁移的索引列表，多个索引使用英文逗号（,）分隔。
        index => "kibana_sample_data_logs"
        # 按时间范围查询增量数据，以下配置表示查询最近5分钟的数据。
        query => '{"query":{"range":{"@timestamp":{"gte":"now+8h-5m","lte":"now+8h/m"}}}}'
        # 定时任务，以下配置表示每分钟执行一次。
        schedule => "* * * * *"
        scroll => "5m"
        docinfo=>true
        size => 5000

2、snapshot（含冷热分层场景）

snapshot api 是 ES 提供的一组用于索引数据快速备份及恢复的 api 接口，详细介绍可查看官方文档，利用该组 api 接口，我们可以通过将源 es 集群的数据备份到远程仓库，再在腾讯云es集群上从远程仓库中通过快照恢复的方式来实现跨集群的离线数据迁移。

注意：目标 ES 集群的主版本号（如6.4.3版本中的6为主版本号）要大于等于源 ES 集群的主版本号。1.x 版本的集群创建的快照不能在 5.x 版本中恢复。例如客户的源端集群版本号为6.4.3，则腾讯云ES集群的版本需>= 6.4.3。

快照版本关系说明

适用场景：

snapshot 迁移适用于集群规模较大、索引个数比较多的场景。

迁移步骤

1. 自建ES集群安装COS插件，并重启集群

创建快照前需要在创建 repository 仓库。一个repository仓库可以存储多个数据快照文件。repository 主要有如下几种类型：

fs：共享文件系统，将快照文件存放于文件系统中；

url：指定文件系统的 URL 路径，支持协议：http、https、ftp、file、jar；

s3：AWS S3 对象存储，快照存放于 S3 中，以插件形式支持，安装插件 repository-s3；

hdfs：快照存放于 hdfs 中，以插件形式支持，安装插件 repository-hdfs；

cos：快照存放于腾讯云 COS 对象存储中，以插件形式支持，安装插件 elasticsearch-repository-cos。

其中COS插件的github地址：https://github.com/tencentyun/elasticsearch-repository-cos

从客户自建 ES 集群中迁移到腾讯云 ES，我们可以直接使用 COS 进行迁移，如果客户集群没有安装过 COS 插件，则需要先安装 COS 插件，最后重启集群：

/$you_elasticsearch_dir/bin/elasticsearch-plugin install file:///$you_plugin_path/elasticsearch-cos-x.x.zip

2. 自建ES集群创建仓库

调用 snapshot 的 api 创建 repository 仓库

获取 access 秘钥信息：访问密钥 - 控制台 (tencent.com)

PUT _snapshot/my_cos_backup
{
    "type": "cos",
    "settings": {
        "app_id": "xxxxxxx",
        "access_key_id": "xxxxxx",
        "access_key_secret": "xxxxxxx",
        "bucket": "xxxxxx",
        "region": "ap-guangzhou",
        "compress": true,
        "chunk_size": "500mb",
        "base_path": "/",
        "max_restore_bytes_per_sec":"40m",
        "max_snapshot_bytes_per_sec":"40m"
    }
}

3. 源端ES集群执行创建快照

将源端 ES 集群的索引数据备份到 COS 的 my_cos_backup 仓库下，命名为：snapshot_1

PUT _snapshot/my_cos_backup/snapshot_1
{
  "indices":"*,-.*,-ilm-*"
}

执行完上面的命令后会立即返回，及数据的备份是异步执行的，我们可以通过如下的命令查看快照执行的状态：

GET _snapshot/my_cos_backup/snapshot_1

返回格式如下，其中state若返回的是 "SUCESS"，则表示快照备份完成。若是 "IN_PROCESS"，则表示快照还在备份中。

{
    "snapshots": [
        {
            "snapshot": "snapshot_1",
            "uuid": "zUSugNiGR-OzH0CCcgcLmQ",
            "version_id": 5060499,
            "version": "5.6.4",
            "indices": [
                "index_1",
                "index_2"
            ],
            "state": "SUCCESS",
            "start_time": "2018-05-04T11:44:15.975Z",
            "start_time_in_millis": 1525434255975,
            "end_time": "2018-05-04T11:45:29.395Z",
            "end_time_in_millis": 1525434329395,
            "duration_in_millis": 73420,
            "failures": [],
            "shards": {
                "total": 3,
                "failed": 0,
                "successful": 3
            }
        }
    ]
}

上面备份的是整个索引，如果我们只想将部分索引迁移到云上，也可以通过下面的命令：

PUT _snapshot/my_cos_backup/snapshot_1
{
    "indices": "index_1,index_2"
}

4. 在腾讯云ES集群上创建快照仓库

获取access秘钥信息：访问密钥 - 控制台 (tencent.com)

PUT _snapshot/my_cos_backup
{
    "type": "cos",
    "settings": {
        "app_id": "xxxxxxx",
        "access_key_id": "xxxxxx",
        "access_key_secret": "xxxxxxx",
        "bucket": "xxxxxx",
        "region": "ap-guangzhou",
        "compress": true,
        "chunk_size": "500mb",
        "base_path": "/",
        "max_restore_bytes_per_sec":"40m",
        "max_snapshot_bytes_per_sec":"40m"
    }
}

● app_id：腾讯云账号 APPID。

● access_key_id：腾讯云 API 密钥 SecretId。

● access_key_secret：腾讯云 API 密钥 SecretKey。

bucket：COS Bucket 名字，名字不能带 -{appId}后缀。

● region：COS Bucket 地域，此地域必须与 ES 集群为同一地域。地域编码可参考地域和可用区。

● base_path：备份目录。

5. 云 ES 集群上执行恢复命令

从 COS 仓库中恢复快照：

1）elasticsearch版本 >= 7.10

指定热节点恢复：

POST _snapshot/my_cos_backup/snapshot_1/_restore
{
  "indices":"*,-.*,-ilm-*",
  "index_settings": {
    "index.routing.allocation.include._tier_preference": "data_hot,data_warm,data_cold"
}

指定冷节点恢复：

POST _snapshot/my_cos_backup/snapshot_1/_restore
{
  "indices":"*,-.*,-ilm-*",
  "index_settings": {
    "index.routing.allocation.include._tier_preference": "data_warm,data_hot,data_cold"
  }
}

2）elasticsearch版本 < 7.10

指定热节点恢复：

POST _snapshot/my_cos_backup/snapshot_1/_restore
{
  "indices":"*,-.*,-ilm-*",
  "index_settings": {
    "index.routing.allocation.require.temperature": "hot"
}

指定冷节点恢复：

POST _snapshot/my_cos_backup/snapshot_1/_restore
{
  "indices":"*,-.*,-ilm-*",
  "index_settings": {
    "index.routing.allocation.require.temperature": "warm"
  }
}

上面这一步就是将 COS 中刚备份的快照数据恢复到云上的ES集群中，执行这一步的前提是需要先在云上的 ES 集群中创建仓库，地址和源端集群上一致。

从快照中恢复指定索引：

POST /_snapshot/my_cos_backup/snapshot_1/_restore
{
 "indices": "index_1",
 "index_settings": {
   "index.routing.allocation.include._tier_preference": "data_warm,data_hot,data_cold"
}

6. 查看快照状态

快照恢复状态

GET index_1/_recovery

7. 增量备份/恢复

增量备份：

增量备份就是在原有的快照仓库里，新建一个快照，api完全一致，比如第一次全量备份的快照为 _snapshot/my_cos_backup/snapshot_1，第二备份的快照 _snapshot/my_cos_backup/snapshot_2 就是基于第一次全量的增量备份。

增量恢复：

在目标集群的增量恢复，需要先_close目标需要恢复的索引，比如需要恢复names_index这个索引，则需要在目标端进行 _close 的操作，关闭索引后即可进行增量恢复。

关闭全量索引：

POST *,-.*/_close

如果报错不允许模糊匹配，可以开启一下通配权限：

PUT _cluster/settings
{
  "transient": {
    "action.destructive_requires_name": "false"
  }
}

总结

1. Logstash 适用于集群规模较大的场景。这两种迁移方案都需要实行迁移的机器能够同时访问到源ES集群和腾讯云 ES 集群。并且这两种方案都不需要源 ES 集群重启。但是在迁移过程中迁移工具退出或者执行迁移的服务器宕机，则会从头开始迁移，因此需要先将已经迁移的部分数据清空。迁移完成后通过比较索引的 doc 数量来确定索引是否全部迁移成功；

2. snapshot 适用于集群规模大、索引数量多、对迁移时效有较高要求的场景。但是由于 snapshot 在迁移前需要源ES集群的每个节点都配置 repository 仓库地址，因此集群需要重启，这可能会影响源 ES 集群的查询性能和稳定性。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

Elasticsearch Service