腾讯云Elasticsearch ServiceElasticsearch实践与分析@IT·大数据

记一次在线跨集群迁移ES数据

2019-12-03  本文已影响0人  bellengao

背景

业务所有的服务器日志都是通过filebeat进行采集,然后写入到一个公共的ES集群中。因为当前使用的集群无法继续扩容了并且版本也较低(5.6.4), 所以需要把集群迁移到一个新的规模更大的集群,并且升级一下ES的版本,升级到6.4.3.

方案制定

迁移的需求是:

  1. 日志数据不能停止写入到ES
  2. 日志查询不受影响,延迟不能超过1分钟

参考之前写的关于数据迁移的文章Elasticsearch数据迁移与集群容灾,制定出的迁移方案为:

  1. 先使用logstash或者snapshot全量同步一次数据到新集群中
  2. 使用logstash追平当天的日志索引后,查询入口切换到新的ES集群
  3. 日志写入入口切换到新的ES集群

实施步骤

1. 新建6.4.3版本ES集群

如果旧的日志集群处理日志时使用了ingest pipeline, 新的集群也需要同样配置pipeline。

2.使用logstash/snapshot全同步数据

如果数据规模较小,比如几十GB, 则可以使用logstash进行全量同步, logstash配置文件如下:

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "*"
        docinfo => true
        size => 5000
        scroll => "5m"
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
        pipeline => "timezone-pipeline"
    }
}

实施过程中遇到的两个问题:

    "description": "timezone-pipeline",
    "processors": [
      {
        "date": {
          "field": "@timestamp",
          "formats": [
            "ISO8601"
          ],
          "timezone": "Asia/Shanghai",
          "ignore_failure": true
        }
      }
    ],
    "on_failure": [
      {
        "set": {
          "field": "error",
          "value": "{{ _ingest.on_failure_message }}"
        }
      }
    ]

如果数据规模较大,几百GB以上,则可以使用snapshot进行一次全量的迁移,速度较快。

3. 记录新集群中当天索引中数据的最新时间戳

存量的旧的索引不会再写入了,而当天的索引还在持续写入,在步骤2的全量同步数据完成之后(logstash执行完毕后会自动终止进程), 需要查询出当天索引的数据中已经同步完成的最新的时间戳,在新的集群中执行查询:

GET es-runlog-2019-11-20/_search
{
  "query": {
    "match_all": {}
  },
  "size": 1,
  "sort": [
    {
      "@timestamp": "desc"
    }
  ]
}

记执行上述查询获取到的时间戳为start.

4.增量迁移当天的索引

使用logstash增量迁移当天的索引,logstash配置如下:

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "es-runlog-2019.11.20"
        query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
        size => 5000
        scroll => "5m"
        docinfo => true
        schedule => "* * * * *" #定时任务,每分钟执行一次
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
        pipeline => "timezone-pipeline"
    }
}

上述配置,每分钟执行一次,从源集群中拉取5分钟前到当前分钟的所有数据,同步到新的集群中;因为查询的粒度为分钟,所以每次执行定时任务查询时会有一部分重叠的数据,所以需要在output中配置document_id参数避免重复写入到新集群中。

实施过程中遇到的问题有:

  1. 用于运行logstash的机器的规格要比较大,因为logstash比较消耗内存和cpu,机器性能不够,很可能出现数据同步延迟增大
  2. 可以通过比较新旧集群当天的索引每分钟doc数据量,判断同步的延迟情况,如果延迟较大,可以通过调整logstash配置或者使用更大的机器运行logstash确保同步过程顺利进行

5. 记录开始迁移的时间

在新的集群中执行以下查询,记录开始进行增量迁移的时间戳:

GET es-runlog-2019-11-20/_search

{
  "query": {
    "range": {
      "@timestamp": {
        "gt": "{start}"
      }
    }
  },
  "size": 1,
  "sort": [
    {
      "@timestamp": "asc"
    }
  ]
}

记获取到的时间戳为end.

6. 追平start和end之间的数据

使用logstash从源集群中获取start和end之间的日志数据,同步到新集群中,配置文件如下:

input {
    elasticsearch {
        hosts => "1.1.1.1:9200"
        index => "es-runlog-2019.11.20"
        query => '{"query":{"range":{"@timestamp":{"gte":"start","lt":"end"}}}}'
        docinfo => true
      }
}
filter {
     mutate {
   remove_field => ["source", "@version"]
 }
}
output {
    elasticsearch {
        hosts => ["http://2.2.2.2:9200"]
        index => "%{[@metadata][_index]}"
         document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
         pipeline => "timezone-pipeline"
    }
}

7. 持续观察数据同步过程是否稳定

待步骤6的数据追平过程结束之后,需要持续观察步骤5的增量迁移数据的情况是否稳定,待一段时间,比如几个小时之后,仍然可以稳定的进行同步,此时可以把日志的查询入口切换到新集群中,之后再把数据写入入切换到新集群中,至此,一次在线跨集群迁移数据实施过程完毕。

上一篇下一篇

猜你喜欢

热点阅读