通过Logstash迁移自建Elasticsearch数据

全量数据迁移

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

1. 创建Logstash配置文件fulldata.conf，用于配置全量数据迁移任务。具体如下：

   input{
     elasticsearch{
       # 源ES地址
       hosts =>  ["http://<host>:<port>"]
       
       # 访问使用的用户名&密码
       user => "changeme"
       password => "changeme"
       
       # 源index名，支持逗号分隔的多个index名和模糊匹配
       index => "geonames"
     
       # 查询所有数据并同步输出源
       query => '{"query":{"match_all":{}}}'
   
       # 默认为false，不会读出_id等元数据信息
       docinfo=>true
     }
   }
   
   filter {
     # 去掉Logstash附加字段
     mutate {
       remove_field => ["@timestamp", "@version"]
     }
   }
   
   output{
     elasticsearch{
       # Lindorm连接地址
       hosts => ["http://<lindorm-address>"]
     
       # 访问使用的用户名&密码
       user => "changeme"
       password => "changeme"
     
       index => "geonames"
       # 指定数据写入时，保持id为原来的值，如无业务需要可以删去
       document_id => "%{[@metadata][_id]}"
   
       # 关闭Logstash自带的模板
       manage_template => false
     }
   }

   参数说明

   配置项	参数	说明
   input	hosts	自建Elasticsearch集群的IP和端口。
   	user	自建Elasticsearch集群的用户名。非必填项，请根据实际情况填写。
   	password	自建Elasticsearch集群的密码。非必填项，请根据实际情况填写。
   	index	自建Elasticsearch集群中待迁移的索引名。
   	query	迁移数据时使用的查询条件。示例代码中match_all表示查询索引中的所有文档，即将数据源索引中的所有文档都迁移到目标索引中。
   output	hosts	搜索引擎（Elasticsearch兼容版本）的Elasticsearch兼容连接地址。如何获取，请参见Elasticsearch兼容版本。 重要 如果您的Logstash服务部署在ECS实例，建议您通过专有网络访问Lindorm实例，可获得更高的安全性和更低的网络延迟。 如果您的Logstash服务部署在本地，在通过公网连接Lindorm实例前，需在控制台开通公网地址。开通方式：在控制台的左侧导航栏，选择数据库连接，单击搜索引擎页签，在页签右上角单击开通公网地址。 通过专有网络访问Lindorm实例，请填写Elasticsearch兼容地址对应的专有网络地址。通过公网访问Lindorm实例，请填写Elasticsearch兼容地址对应的公网地址。
   	user	访问搜索引擎（Elasticsearch兼容版本）的用户名和密码。 默认用户名和密码的获取方式：在控制台的左侧导航栏，选择数据库连接，单击搜索引擎页签，在搜索引擎页签可获取。
   	password
   	index	Lindorm搜索引擎中创建的索引名（目标索引）。

2. 指定fulldata.conf为任务配置文件并启动Logstash服务，等待数据迁移完成。数据迁移完成后，Logstash服务会自动停止。

   cd logstash-7.10.0
   bin/logstash -f <fulldata.conf所在路径>

增量数据迁移

如果您自建的ES索引存在数据更新或写入了新数据，且没有数据被删除，则可以通过增量数据迁移的方式完成数据迁移任务。您可以根据ES索引的mapping参数设置，选择不同的方案进行数据迁移。

• 如果您的数据中不包含数据更新时间字段，那么您需要在您数据写入的业务代码中添加写入数据更新时间的逻辑。对不包含数据更新时间字段的历史数据应用全量数据迁移方案，对包含数据更新时间字段的增量数据应用滚动迁移的方案。本文使用的Logstash增量数据迁移任务配置文件history.conf如下：

  input{
    elasticsearch{
      hosts =>  ["http://<host>:<port>"]
      user => "changeme"
      password => "changeme"
      index => "geonames"
      query => '{"query":{"bool":{"must_not":{"exists":{"field":"updateTimeField"}}}}}'
      docinfo=>true
    }
  }
  filter {
    mutate {
      remove_field => ["@timestamp", "@version"]
    }
  }
  output{
    elasticsearch{
      hosts => ["http://<lindorm-address>"]
      user => "changeme"
      password => "changeme"
      index => "geonames"
      document_id => "%{[@metadata][_id]}"
      manage_template => false
    }
  }

• 如果您的数据包含数据更新时间字段，则可以基于数据更新时间字段，使用滚动迁移的方式来完成数据迁移任务。本文使用的Logstash增量数据迁移配置文件increment.conf和任务滚动迁移脚本如下。

  • 配置文件increment.conf。

    input{
      elasticsearch{
        hosts =>  ["<自建ES集群的连接地址>:<自建ES集群的端口>"]
        user => "<自建ES集群的用户名>"
        password => "<自建ES集群的密码>"
        index => "<自建ES集群的数据源索引>"
        query => '{"query":{"range":{"updateTimestampField":{"gte":"${TMP_LAST}","lt":"${TMP_CURR}"}}}}'
        docinfo=>true
      }
    }
    filter {
      mutate {
        remove_field => ["@timestamp", "@version"]
      }
    }
    output{
      elasticsearch{
        hosts => ["http://<lindorm-address>"]
        user => "changeme"
        password => "changeme"
        index => "geonames"
        document_id => "%{[@metadata][_id]}"
        manage_template => false
      }
    }

  • 任务滚动迁移脚本。

    #!/bin/bash
    
    unset TMP_LAST
    unset TMP_CURR
    
    # 滚动迁移执行间隔，默认60s
    sleepInterval=60
    
    # 默认转化为毫秒（ms），需要根据业务字段的单位进行调整
    export TMP_LAST=0
    i=1
    while true
    do
      echo "开始第 ${i} 次Logstash迁移数据任务..."
    
      # 默认转化为毫秒（ms），需要根据业务字段的单位进行调整
      export TMP_CURR=$(($(date +%s%N)/1000000))
      <path-to-logstash>/bin/logstash -f <path-to-increment.conf>
    
      echo "完成第 ${i} 次Logstash迁移数据任务..."
      echo "本次迁移数据的数据更新时间范围：${TMP_LAST} 到 ${TMP_CURR}"
      i=$(( $i + 1 ))
      export TMP_LAST=${TMP_CURR}
      sleep ${sleepInterval}
    done

    以上示例代码在increment.conf配置文件中使用了环境变量，结合滚动迁移脚本，持续迁移一段时间窗口内更新的数据。更多有关在Logstash任务配置中使用环境变量的说明，请参见Using environment variables。

    重要

    由于无法保证数据写入请求到达Lindorm搜索引擎服务端的顺序，因此服务端处理请求的顺序也不确定。如果同时执行全量和增量数据迁移任务，可能会出现历史数据将更新后的数据覆盖的情况。因此，建议您先执行历史全量数据迁移任务，待任务执行结束后，再启动增量数据滚动迁移任务。