通过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索引存在数据更新或是写入了新数据，且没有数据被删除，则可以通过增量数据迁移的方式完成数据迁移任务。您可以根据ES索引的mappings，选择不同的方案进行数据迁移。

• 如果您的数据包含数据更新时间字段，则可以基于数据更新时间字段，仅使用滚动迁移的方式来完成数据迁移任务。

  滚动迁移示例代码如下：

  #!/bin/bash
  # 滚动迁移脚本
  
  # Reindex执行间隔，默认60s
  sleepInterval=60
  
  # 默认转化为ms，需要根据业务字段的单位调整
  lastReindexFinish=0
  i=1
  while true
  do
    echo "开始第 ${i} 次Reindex任务..."
    # 默认转化为ms，需要根据业务字段的单位调整
    currReindexStart=$(($(date +%s%N)/1000000))
    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搜索引擎中创建的目标索引>"
    }
  }'

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