ElasticSearch7.3学习(十四)----生产环境实时重建索引
1、实时重建索引
在实际的生产环境中，一个field的设置是不能被修改的，如果要修改一个Field，那么应该重新按照新的mapping，建立一个index，然后将数据批量查询出来，重新用bulk api写入index中。
批量查询的时候，建议采用scroll api，并且采用多线程并发的方式来reindex数据。例如说每次scoll就查询指定日期的一段数据，交给一个线程即可。
(1)一开始，依靠dynamic mapping，插入数据，但是不小心有些数据是2019-09-10这种日期格式的，所以title这种field被自动映射为了date类型，实际上它应该是string类型的。
首先插入以下数据
PUT /my_index/_doc/1
{
"title": "2019-09-10"
}
PUT /my_index/_doc/2
{
"title": "2019-09-11"
}
（2）当后期向索引中加入string类型的title值的时候，就会报错
PUT /my_index/_doc/3
{
"title": "my first article"
}
报错
{
"error": {
"root_cause": [
{
"type": "mapper_parsing_exception",
"reason": "failed to parse field [title] of type [date] in document with id '3'. Preview of field's value: 'my first article'"
}
],
"type": "mapper_parsing_exception",
"reason": "failed to parse field [title] of type [date] in document with id '3'. Preview of field's value: 'my first article'",
"caused_by": {
"type": "illegal_argument_exception",
"reason": "failed to parse date field [my first article] with format [strict_date_optional_time||epoch_millis]",
"caused_by": {
"type": "date_time_parse_exception",
"reason": "Failed to parse with all enclosed parsers"
}
}
},
"status": 400
}
（3）如果此时想修改title的类型，是不可能的
PUT /my_index/_mapping
{
"properties": {
"title": {
"type": "text"
}
}
}
报错
{
"error": {
"root_cause": [
{
"type": "illegal_argument_exception",
"reason": "mapper [title] of different type, current_type [date], merged_type [text]"
}
],
"type": "illegal_argument_exception",
"reason": "mapper [title] of different type, current_type [date], merged_type [text]"
},
"status": 400
}
（4）此时，唯一的办法，就是进行reindex，也就是说，重新建立一个索引，将旧索引的数据查询出来，再导入新索引。
（5）如果说旧索引的名字，是old_index，新索引的名字是new_index，终端java应用，已经在使用old_index在操作了，难道还要去停止java应用，修改使用的index为new_index，才重新启动java应用吗？这个过程中，就会导致java应用停机，可用性降低。
（6）所以说，给java应用一个别名，这个别名是指向旧索引的，java应用先用着，java应用先用prod_index来操作，此时实际指向的是旧的my_index
PUT /my_index/_alias/prod_index
（7）查看别名，会发现my_index已经存在一个别名prod_index了。
GET my_index/_alias
（8）新建一个index，调整其title的类型为string
PUT /my_index_new
{
"mappings": {
"properties": {
"title": {
"type": "text"
}
}
}
}
（9）使用scroll api将数据批量查询出来
GET /my_index/_search?scroll=1m
{
"query": {
"match_all": {}
},
"size": 1
}
返回
{
"_scroll_id" : "DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAARUMWQWx5bzRmTW9TeUNpNmVvN0E2dF9YQQ==",
"took" : 4,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 2,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "my_index",
"_type" : "_doc",
"_id" : "1",
"_score" : 1.0,
"_source" : {
"title" : "2019-09-10"
}
}
]
}
}
（9）采用bulk api将scoll查出来的一批数据，批量写入新索引
POST /_bulk
{"index":{"_index":"my_index_new","_id":"1"}}
{"title":"2019-09-10"}
（10）反复循环8~9，查询一批又一批的数据出来，采取bulk api将每一批数据批量写入新索引
（11）将my_index索引的别名prod_index切换到my_index_new上去，java应用会直接通过index别名使用新的索引中的数据，java应用程序不需要停机，零提交，高可用
POST /_aliases
{
"actions": [
{
"remove": {
"index": "my_index",
"alias": "prod_index"
}
},
{
"add": {
"index": "my_index_new",
"alias": "prod_index"
}
}
]
}
（12）直接通过prod_index别名来查询，是否ok
GET prod_index/_search
可以看到能够查询到新索引my_index_new的数据了
{
"took" : 1117,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 1,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "my_index_new",
"_type" : "_doc",
"_id" : "1",
"_score" : 1.0,
"_source" : {
"title" : "2019-09-10"
}
}
]
}
}
2、总结：
基于alias对client透明切换index
PUT /my_index_v1/_alias/my_index
client对my_index进行操作
reindex操作，完成之后，切换v1到v2
POST /_aliases
{
* "actions": [
* * * { "remove": { "index": "my_index_v1", "alias": "my_index" }},
* * * { "add": * { "index": "my_index_v2", "alias": "my_index" }}
* ]
}
本文来自博客园，作者：|旧市拾荒|，转载请注明原文链接：https://www.cnblogs.com/xiaoyh/p/16028045.html
posted @ 2022-03-26 23:01 |旧市拾荒| 阅读( 394) 评论( 0) 编辑 收藏 举报
版权声明：本文不是「本站」原创文章，版权归原作者所有 | 原文地址: