Elasticsearch索引中数据的增删改查与并发控制

Elasticsearch提供了丰富的REST API接口，使得用户能够便捷地对索引映射中的数据进行增删改查操作。本文将通过示例演示这些操作方法，同时介绍如何利用乐观锁进行并发控制，以及如何实现索引数据的批量写入。

使用REST端点对索引映射中的数据进行增删改查

首先，创建一个名为person的索引，并定义其映射结构：

PUT person
{
  "mappings": {
    "properties": {
      "id": {
        "type": "integer"
      },
      "sex": {
        "type": "boolean"
      },
      "name": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "born": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss"
      },
      "location": {
        "type": "geo_point"
      }
    }
  }
}
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
29

接下来，执行以下操作：

写入一条数据

POST person/_doc/1
{
  "id": "1",
  "sex": true,
  "name": "张三",
  "born": "2020-09-18 00:02:20",
  "location": {
    "lat": 41.12,
    "lon": -71.34
  }
}
1
2
3
4
5
6
7
8
9
10
11

查询第一条数据

GET person/_doc/1
1

修改数据

POST person/_update/1
{
  "doc": {
    "sex": false,
    "born": "2020-02-24 00:02:20"
  }
}
1
2
3
4
5
6
7

删除数据

DELETE person/_doc/1
1

插入数据（覆盖原有文档）

POST person/_doc/1
{
  "id": "1",
  "sex": true,
  "name": "张三",
  "born": "2020-09-18 00:02:20",
  "location": {
    "lat": 41.12,
    "lon": -71.34
  }
}
1
2
3
4
5
6
7
8
9
10
11

查询结果如下：

{
  "_index" : "person",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 4,
  "_seq_no" : 9,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "id" : "1",
    "sex" : false,
    "name" : "张三",
    "born" : "2020-09-11 00:02:20",
    "location" : {
      "lat" : 41.12,
      "lon" : -71.34
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

使用乐观锁进行并发控制

在高并发环境下，可能会有多个请求同时尝试修改同一份数据。为了保证数据一致性，可以利用Elasticsearch提供的乐观锁机制进行并发控制。在更新数据时，传递if_seq_no和if_primary_term参数，确保只有当当前版本与预期版本一致时才执行更新：

PUT person/_doc/1?if_seq_no=9&if_primary_term=1
{
  "id": "1",
  "sex": false,
  "name": "张三",
  "born": "2020-09-11 00:02:20",
  "location": {
    "lat": 41.12,
    "lon": -71.34
  }
}
1
2
3
4
5
6
7
8
9
10
11

索引数据的批量写入

Elasticsearch支持一次请求内执行多条数据的增删改操作，极大地提高了数据写入效率。批量写入的操作类型包括index、create、update、delete。

批量添加或更新数据

POST person/_bulk
{"index":{"_id":"3"}}
{"id":"3","name":"王五","sex":true,"born":"2020-09-14 00:02:20","location":{"lat":11.12,"lon":-71.34}}
{"index":{"_id":"4"}}
{"id":"4","name":"李四","sex":false,"born":"2020-10-14 00:02:20", "location":{"lat":11.12,"lon":-71.34}}
{"index":{"_id":"5"}}
{"id":"5","name":"黄六","sex":false,"born":"2020-11-14 00:02:20", "location":{"lat":11.12,"lon":-71.34}}
1
2
3
4
5
6
7

混合操作示例

POST person/_bulk
{"index":{"_id":"2"}}
{"id":"2","name":"赵二","sex":true,"born":"2020-09-14 00:02:20","location":{"lat":11.12,"lon":-71.34}}
{"create":{"_id":"4"}}
{"id":"4","name":"李四","sex":false,"born":"2020-10-14 00:02:20", "location":{"lat":11.12,"lon":-71.34}}
{"update":{"_id":"5"}}
{ "doc" : {"sex" : "false","born" : "2020-01-01 00:02:20"} }
{"delete":{"_id":"5"}}
1
2
3
4
5
6
7
8

索引重建。

首先，创建一个新的索引newperson，并设置其分片数和副本数：

PUT newperson
{
  "settings": {
    "number_of_shards": "5",
    "number_of_replicas": "1"
  }
}
1
2
3
4
5
6
7

接下来，使用_reindex API将原索引person中的所有数据迁移到新索引newperson：

POST _reindex
{
  "source": {
    "index": "person"
  },
  "dest": {
    "index": "newperson"
  }
}
1
2
3
4
5
6
7
8
9

以上操作将按照原索引person的映射结构和数据顺序，将所有文档复制到新索引newperson中。完成迁移后，新索引将拥有与原索引相同的文档集合，但可以根据需要拥有不同的配置、映射或元数据。

值得注意的是，_reindex操作可能消耗大量系统资源，尤其是当源索引数据量较大时。因此，在执行此操作前应确保集群有足够的资源，并尽量选择低峰时段进行。此外，如果源索引在重建期间有新的数据写入，这些数据不会被包含在重建后的索引中。因此，可能需要暂停相关应用程序的写入操作，或者在重建完成后重新同步新增数据。

总结来说，Elasticsearch提供了丰富的REST API接口，使得用户能够方便地对索引中的数据进行增删改查操作。通过使用乐观锁机制，可以有效防止并发冲突，保障数据一致性。此外，批量写入功能极大地提升了数据写入的效率，适用于大规模数据导入或频繁更新的场景。