Elasticsearch-35-使用scroll+bulk+索引别名实现零停机重建索引

发表于 | 分类于

场景

如果我们一开始新建了一个索引,并且依靠dynamic mapping,这个时候插入一条数据是2018-01-01这种格式的,这field就会被自动映射成了date类型,但是其实他应该是个string类型的,这时候应该怎么做呢?

解决方案

一个field的设置是不能被修改的,如果要修改一个field,那么应该重新按照新的mapping来创建一个index,然后将旧的index中的数据查询出来,用_bulk api批量插入到新的索引中去

批量查询的时候,建议采用scroll api,采用多线程并发的方式来reindex数据.

案例

我们先插入一条数据如下:

1
2
3
4
PUT /old_my_index/my_type/1
{
  "title":"2017-01-01"
}

然后获取这个index的mapping 

1
2
3
4
5
6
7
8
9
10
11
12
13
{
  "old_my_index": {
    "mappings": {
      "my_type": {
        "properties": {
          "title": {
            "type": "date"
          }
        }
      }
    }
  }
}

可以看到title已经被映射成了date类型,这时 如果我们在添加一个字符串的值是添加不进去的. 而且如果想修改这个field的类型也是不可能的

此时唯一的办法就是进行reindex,也就是说重新建立一个索引,将旧索引中的数据查询出来,导入新索引

这里可能会有一个问题,旧的索引名称是old_my_index,假如新的索引名称是new_my_index, 这时候已经有一个java应用在使用old_my_index在操作了, 那么这时候是不是要先停止应用,修改索引,然后重启呢? 这样的话会导致java应用停机,降低可用性

针对上面的问题呢,我们可以先给java应用一个旧索引的别名, java用的只是一个别名,指向旧的索引 

1
PUT /old_my_index/_alias/my_index

执行上面的代码,就是给了old_my_index一个别名(my_index),然后我们新建一个索引,将title这个field调整为string类型的 

1
2
3
4
5
6
7
8
9
10
11
12
PUT /new_my_index
{
  "mappings": {
    "my_type":{
      "properties": {
        "title":{
          "type": "text"
        }
      }
    }
  }
}

新建完成以后,用scroll api从旧的索引中查询数据

1
2
3
4
5
6
7
8
GET old_my_index/my_type/_search?scroll=1m
{
  "query": {
    "match_all": {}
  },
  "sort": ["_doc"],
  "size": 1
}

返回值:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
{
  "_scroll_id": "DnF1ZXJ5VGhlbkZldGNoBQAAAAAAAAHVFmY1N3VWOTF4U19HUlRRUzJIbzgxcmcAAAAAAAAB1xZmNTd1VjkxeFNfR1JUUVMySG84MXJnAAAAAAAAAdQWZjU3dVY5MXhTX0dSVFFTMkhvODFyZwAAAAAAAAHYFmY1N3VWOTF4U19HUlRRUzJIbzgxcmcAAAAAAAAB1hZmNTd1VjkxeFNfR1JUUVMySG84MXJn",
  "took": 3,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": null,
    "hits": [
      {
        "_index": "old_my_index",
        "_type": "my_type",
        "_id": "1",
        "_score": null,
        "_source": {
          "title": "2017-01-01"
        },
        "sort": [
          0
        ]
      }
    ]
  }
}

查询出来以后用bulk api将scroll查出来的一批数据,批量写入新的索引. 

1
2
3
POST /_bulk
{"index":{"_index":"new_my_index","_type":"my_type","_id":"1"}}
{"title":"2017-01-01"}

重复循环scroll查询和bulk批量插入,直到所有的数据都添加到了新的索引中

添加完成后,将别名切换到新的索引上去,这样的话java应用就直接通过别名使用新的索引中的数据了,不需要停机重启,高可用 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
POST /_aliases
{
  "actions": [
    {
      "remove": {       // 把别名从旧的索引上先移除
        "index": "old_my_index",
        "alias": "my_index"
      }
    },
    {
      "add": {
        "index": "new_my_index", // 将别名指向新的索引
        "alias": "my_index"
      }
    }
  ]
}

总结

总体来说,就是最开始就给索引一个别名去让客户端去使用,然后如果要切换索引的话,就先建一个索引,然后查询旧的索引数据,将数据插入到新的索引中,完成后将别名指向新的索引,就实现了零停机重建索引