通过Reindex API迁移自建Elasticsearch数据

全量数据迁移

如果您的自建ES集群中的数据源索引不再有任何变化，即没有文档的增、删、改，则可以通过全量数据迁移的方式将数据源索引中的所有文档迁移至Lindorm搜索引擎。通过在Lindorm搜索引擎上执行Reindex请求，即可触发全量数据迁移任务。

curl -XPOST "http://<Lindorm搜索引擎的连接地址>/_reindex" -u <username>:<password> -H'Content-Type:application/json' -d'{
  "source": {
    "remote": {
      "host": "http://<host>:<port>",
      "username": "changeme",
      "password": "changeme"
    },
    "index": "geonames",
    "query": {
      "match_all": {}
    }
  },
  "dest": {
    "index": "geonames"
  }
}'

参数说明

如果您需要异步执行Reindex请求，可以指定wait_for_completion参数的值为false，如下所示：

curl -XPOST "http://<Lindorm搜索引擎的连接地址>/_reindex?wait_for_completion=false" -u <username>:<password>

Lindorm搜索引擎将返回一个字符串，表示Reindex请求处理对应的任务ID，您可以通过Task management API查看任务的执行情况。更多Reindex API使用详情，请参见Reindex API。

增量数据迁移

如果您自建的ES索引存在数据更新或是写入了新数据，且没有数据被删除，则可以通过增量数据迁移的方式完成数据迁移任务。

建议您基于数据更新时间字段，仅使用滚动迁移的方式来完成数据迁移任务，在滚动迁移任务第一次执行时，会将历史数据一并迁移至Lindorm搜索引擎中。

滚动迁移示例代码如下：

#!/bin/bash
# 滚动迁移脚本

# Reindex执行间隔，默认30s
sleepInterval=30
# 设置一个稍大于源端的refresh interval的值，单位s，此处默认源端是15s
refreshInterval=16

# 默认转化为ms，需要根据业务字段的单位调整
lastReindexFinish=0
i=1
while true
do
  echo "开始第 ${i} 次Reindex任务..."
  # 排除数据写入源端，refresh之前查不到的影响
  # 默认转化为ms，需要根据业务字段的单位调整
  currReindexStart=$((($(date +%s%N)/1000000) - (${refreshInterval} * 1000)))
  curl -XPOST "http://<目的Lindorm搜索连接地址>/_reindex" -u <目的Lindorm搜索用户名>:<目的Lindorm搜索密码> -H'Content-Type: application/json' -d'{
    "source": {
      "remote": {
        "host": "<源ES集群连接地址>",
        "username": "<源ES集群用户名>",
        "password": "<源ES集群密码>"
      },
      "index": "<源ES集群数据源索引>",
      "query": {
        "range" : {
          "<数据更新时间字段名>" : {
            "lt" : '${currReindexStart}',
            "gte" : '${lastReindexFinish}'
          }
        }
      }
    },
    "dest": {
      "index": "<目的Lindorm搜索数据目的索引>"
    }
  }'
  echo "完成第 ${i} 次Reindex任务..."
  echo "本次迁移数据的数据更新时间范围：${lastReindexFinish} 到 ${currReindexStart}"
  i=$(( $i + 1 ))
  lastReindexFinish=${currReindexStart}
  sleep ${sleepInterval}
done

综合迁移方案

如果您的历史数据中不包含数据更新时间字段，那么您需要在您数据写入的业务代码中添加写入数据更新时间的逻辑。对不包含数据更新时间字段的历史数据，使用历史数据迁移脚本进行迁移，对包含数据更新时间字段的增量数据使用滚动迁移方案进行迁移。

当存在数据更新时，建议在历史数据迁移结束后，再启动增量数据迁移，避免通过增量数据迁移写入的新版本数据被历史数据迁移写入的老版本数据覆盖。

历史数据迁移脚本示例如下：

#!/bin/bash
# 历史数据全量迁移脚本

# wait_for_completion可以修改为false，然后通过task management api查看进度
curl -XPOST "http://<Lindorm搜索引擎的连接地址>/_reindex?wait_for_completion=true&pretty" -u <Lindorm搜索引擎的用户名>:<Lindorm搜索引擎的密码> -H'Content-Type:application/json' -d'{
  "source": {
    "remote": {
      "host": "<自建ES集群的连接地址>",
      "username": "<自建ES集群的用户名>",
      "password": "<自建ES集群的密码>"
    },
    "index": "<自建ES集群的数据源索引>",
    "query": {
      "bool": {
        "must_not": {
          "exists": {
            "field": "<数据更新时间字段名>"
          }
        }
      }
    }
  },
  "dest": {
      "index": "<Lindorm搜索引擎中创建的目标索引>"
  }
}'

滚动迁移的示例代码请参见滚动迁移示例代码。