ElasticSearch线上索引重建

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ElasticSearch线上索引重建相关的知识,希望对你有一定的参考价值。

项目背景:

  1.由于项目中存在旧索引设置不合理情况,需要进行索引重建

  2.线上的ElasticSearch由1台扩容到3台,原有的索引需要分片

  例如:

   旧索引 index_user 设置主分片为1,副分片为0,数据没有高可用

GET index_user/_search
{
  "took" : 121,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  }

实现步骤:

  1.新建索引,index_user_v2设置我们所需要的主分片和副分片数量

PUT index_user_v2
{
  "settings": {
    "number_of_replicas": 1,
    "number_of_shards": 5
  }
}

  2.设置索引数据结构,因为新索引和旧索引mapping结构一致,索引可以直接copy旧索引的数据结构;

PUT index_user_v2/t_user/_mappings
{
  "properties": {
    "age": {
      "type": "integer"
    },
    "ageScope": {
      "type": "keyword"
    },
    "birthday": {
      "type": "long"
    },
    "cityId": {
      "type": "integer"
    },
    "cityName": {
      "type": "keyword"
    },
    "countryCode": {
      "type": "integer"
    },
    "countyId": {
      "type": "integer"
    },
    "create_time": {
      "type": "long"
    },
    "dbId": {
      "type": "long"
    },
    "email": {
      "type": "keyword"
    },
    "gameIds": {
      "type": "text",
      "analyzer": "ik_max_word"
    },
    "isCreateServer": {
      "type": "integer"
    },
    "isDelete": {
      "type": "boolean"
    },
    "nickName": {
      "type": "text",
      "analyzer": "ik_smart"
    },
    "nickNamePingYin": {
      "type": "text",
      "analyzer": "pinyin"
    },
    "nnNumber": {
      "type": "long"
    },
    "provinceId": {
      "type": "integer"
    },
    "provinceName": {
      "type": "keyword"
    },
    "sex": {
      "type": "keyword"
    },
    "signature": {
      "type": "keyword"
    },
    "status": {
      "type": "keyword"
    },
    "telNum": {
      "type": "keyword"
    },
    "updae_time": {
      "type": "long"
    },
    "userId": {
      "type": "long"
    },
    "userType": {
      "type": "keyword"
    },
    "userUrl": {
      "type": "keyword"
    },
    "userUrlNn": {
      "type": "keyword"
    },
    "user_id": {
      "type": "long"
    }
  }
}

  3. 执行完步骤1和步骤2之后,在Kibana->Monitoring->Node里面可以看到索引index_user_v2已经被自动分片到三个节点,如图

           

    这里,正式开始索引重建之前,可以将index_user_v2的副分片数量设置为0,减少副分片写入带来的时间损耗

  PUT index_user_v2/_settings
  {
    "settings": {
      "number_of_replicas": 0
    }
  }

   4.执行索引迁移,将index_user上的数据复制到index_user_v2, 同时设置 wait_for_completion=false 表示索引迁移的请求会在后台执行

# 索引迁移
POST /_reindex?wait_for_completion=false
{
  "source": {
    "index": "index_user"
  },
  "dest": {
    "index":"index_user_v2"
  }
}

  执行后,会生成一个taskId  :  例如:Mroifc1NSJq2s7mf38XxmA:1679363718,后续我们可以使用这个taskId去查询这个迁移任务的状态,耗时,以及执行的进度等等

GET _tasks/Mroifc1NSJq2s7mf38XxmA:1679363718
{
  "completed" : true,
  "task" : {
    "node" : "Mroifc1NSJq2s7mf38XxmA",
    "id" : 1679363718,
    "type" : "transport",
    "action" : "indices:data/write/reindex",
    "status" : {
      "total" : 15480531,
      "updated" : 0,
      "created" : 15480531,
      "deleted" : 0,
      "batches" : 15481,
      "version_conflicts" : 0,
      "noops" : 0,
      "retries" : {
        "bulk" : 0,
        "search" : 0
      },
      "throttled_millis" : 0,
      "requests_per_second" : -1.0,
      "throttled_until_millis" : 0
    },
    "description" : "reindex from [index_user] to [index_user_v2]",
    "start_time_in_millis" : 1623316057822,
    "running_time_in_nanos" : 594661905143,
    "cancellable" : true,
    "headers" : { }
  },
  "response" : {
    "took" : 594661,
    "timed_out" : false,
    "total" : 15480531,
    "updated" : 0,
    "created" : 15480531,
    "deleted" : 0,
    "batches" : 15481,
    "version_conflicts" : 0,
    "noops" : 0,
    "retries" : {
      "bulk" : 0,
      "search" : 0
    },
    "throttled" : "0s",
    "throttled_millis" : 0,
    "requests_per_second" : -1.0,
    "throttled_until" : "0s",
    "throttled_until_millis" : 0,
    "failures" : [ ]
  }
}

    5.任务完成后  

    将旧索引index_user的别名index_user_latest 移除

    新索引index_user_v2添加别名index_user_latest

    至此完成全部的索引重建任务

# 别名替换
POST _aliases
{
  "actions": [
    {
      "add": {
        "index": "index_user_v2",
        "alias": "index_user_latest"
      }
    },
    {
      "remove": {
        "index": "index_user",
        "alias": "index_user_latest"
      }
    }
  ]
}

 

 

事后思考:

  1.执行_reindex索引迁移时,会读取当前index_user旧索引的数量 15480602条数据,将这批数据复制到新索引index_user_v2中

  但是实际生产会持续写数据到旧索引index_user中,导致reindex复制的数据,会略小于实际的数据量

  处理方式:该索引的数据是StreamSet实时同步mysql的数据到ElasticSearch中,这里可以将StreamSet停止,记录复制开始的时间,待复制完成后进行数据的增量同步;

  2.这里有个点可以优化,_reindex复制后, wait_for_completion=false 会生成任务,可以将任务Id写入定时任务中,轮训该任务的状态,任务结束后,可以及时通知;

 

 

 

 

  

   

以上是关于ElasticSearch线上索引重建的主要内容,如果未能解决你的问题,请参考以下文章

ElasticSearch实战(二十)-索引重建高级属性

Elasticsearch系列---实战零停机重建索引

Elasticsearch系列---实战零停机重建索引

Elasticsearch系列---实战零停机重建索引

ELK学习总结(4-1)elasticsearch更改mapping(不停服务重建索引)

(57)ElasticSearch重建索引且保证应用程序不用重启