从 ElasticSearch 中删除数据的几种方式

拉斯特11个月前发布

8300

1. 概述

ElasticSearch是一款功能强大的分布式搜索和分析引擎，擅长提取和查询大量数据。然而，有时需要删除数据，无论是出于合规性、存储优化还是数据准确性的原因。

在本教程中，我们探讨了从 ElasticSearch 中删除数据的各种方法，从删除单个文档到管理生产环境中的大规模删除。

2. 删除单个文档

首先，ElasticSearch 提供了几种从索引中删除单个文档的方法。

2.1. 使用删除 API

首先，从 ElasticSearch 中删除单个文档的最简单方法可能是使用 Delete API。当我们知道确切的文档 ID 和索引名称时，此方法是理想的选择：

$ curl -X DELETE "localhost:9200/customers/_doc/1"
{"_index":"customers","_id":"1","_version":3,"result":"deleted","_shards":{"total":2,"successful":1,"failed":0},"_seq_no":20,"_primary_term":1}

在这个例子中， customers是索引的名称， 1是我们要删除的文档的 ID。

当我们执行此命令时，ElasticSearch 会尝试从 customers索引中删除 ID 为1 的文档。随后，如果该文档存在且被成功删除，ElasticSearch 将返回JSON响应，表明操作成功。

2.2. 使用查询删除

另一方面，当我们需要删除符合特定条件的多个文档时，Delete By Query API 更为高效。此方法可以根据查询删除文档，类似于我们搜索文档的方式：

$ curl -X POST "localhost:9200/customers/_delete_by_query" -H 'Content-Type: application/json' -d'
{
  "query": {
    "range": {
      "last_purchase_date": {
        "lt": "now-1y"
      }
    }
  }
}'
{"took":258,"timed_out":false,"total":4,"deleted":4,"batches":1,"version_conflicts":0,"noops":0,"retries":{"bulk":0,"search":0},"throttled_millis":0,"requests_per_second":-1.0,"throttled_until_millis":0,"failures":[]}

让我们分解一下这个例子：

我们向客户索引的 _delete_by_query端点发送一个 POST 请求
我们使用范围查询来查找 last_purchase_date字段距今不到一年的所有文档
now-1y是 ElasticSearch 日期数学表达式，意思是“从当前时间算起一年前”

此查询将删除上次购买时间超过一年的所有客户文档，因此，这是一种根据特定标准删除过时或不相关数据的有效方法。

但是，使用Delete By Query时我们需要注意以下几点：

该操作不是原子的：如果中途失败，一些文档可能已被删除，而其他文档仍保留
对于大型数据集来说，可能会占用大量资源：最好在非高峰时段运行此类操作

我们还可以添加一个 size参数来限制单次操作删除的文档数量。这可以进一步帮助管理集群上的负载。

3.批量删除操作

继续介绍更有效的大规模删除方法，当处理大量文档时，批量操作可以显著提高性能。Bulk API 在单个请求中执行多个删除操作，从而减少网络开销并提高整体效率。

让我们看一个示例，了解如何使用 Python 和 ElasticSearch 客户端的 Bulk API 进行删除：

from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(["http://localhost:9200"])

def generate_actions(inactive_customer_ids):
    for customer_id in inactive_customer_ids:
        yield {
            "_op_type": "delete",
            "_index": "customers",
            "_id": customer_id
        }

inactive_customer_ids = ["3", "5", "8"]

response = helpers.bulk(es, generate_actions(inactive_customer_ids))
print(f"Deleted {response[0]} documents")

首先，我们创建一个 ElasticSearch 客户端实例，连接到本地 ElasticSearch 服务器。然后，我们定义一个生成器函数generate_actions，为每个客户 ID 生成删除操作。之后，我们创建一个非活跃客户 ID 列表。在实际场景中，这样的列表可能来自数据库查询或其他数据源。

随后，我们使用辅助函数.bulk()执行批量删除操作。最后，我们打印已删除的文档数量。

现在，让我们运行脚本：

$ python3 bulk-removal.py 
Deleted 3 documents

Bulk API 比为每个文档发送单独的删除请求更有效，因为它减少了到 ElasticSearch 集群的网络往返次数以及实际内部操作期间的开销。

4. 使用索引操作删除数据

除了文档级操作外，有时我们可能需要删除较大的数据块。在这种情况下，索引级操作可能更有效。

4.1. 删除整个索引

如果我们需要从索引中删除所有数据，则删除整个索引是最快的方法：

$ curl -X DELETE "localhost:9200/customers"
{"acknowledged":true}

此命令删除客户索引及其所有数据。值得注意的是，这是一个非常快的操作，但它也是不可逆的。

当管理基于时间的索引并想要删除旧数据时，此方法很有用。例如，这是删除上个月的日志索引的常用方法。

4.2. 使用别名实现零停机重建索引

对于能够删除数据同时保持可用性的更细致的方法，我们可以使用索引别名。当我们想要从索引中删除数据子集而不产生任何停机时间时，此方法特别有用。

首先，我们为现有索引创建别名：

$ curl -X POST "localhost:9200/_aliases" -H 'Content-Type: application/json' -d'
{
  "actions": [
    { "add": { "index": "customers", "alias": "current_customers" }}
  ]
}'
{"acknowledged":true,"errors":false}

然后，我们使用更新的设置创建一个新的索引：

$ curl -X PUT "localhost:9200/customers_v2" -H 'Content-Type: application/json' -d'
{
  "mappings": {
    "properties": {
      "email": { "type": "keyword" },
      "name": { "type": "text" }
    }
  }
}'
{"acknowledged":true,"shards_acknowledged":true,"index":"customers_v2"}

接下来，我们重新索引数据，排除不活跃的客户：

$ curl -X POST "localhost:9200/_reindex" -H 'Content-Type: application/json' -d'
{
  "source": {
    "index": "customers",
    "query": {
      "bool": {
        "must_not": {
          "term": { "status": "inactive" }
        }
      }
    }
  },
  "dest": {
    "index": "customers_v2"
  }
}'
{"took":251,"timed_out":false,"total":7,"updated":0,"created":7,"deleted":0,"batches":1,"version_conflicts":0,"noops":0,"retries":{"bulk":0,"search":0},"throttled_millis":0,"requests_per_second":-1.0,"throttled_until_millis":0,"failures":[]}

最后，我们将别名切换为新索引：

$ curl -X POST "localhost:9200/_aliases" -H 'Content-Type: application/json' -d'
{
  "actions": [
    { "remove": { "index": "customers", "alias": "current_customers" }},
    { "add":    { "index": "customers_v2", "alias": "current_customers" }}
  ]
}'
{"acknowledged":true,"errors":false}

使用此方法，应用程序可以在整个过程中继续对current_customers别名进行读写 。一旦重新索引完成并切换别名，就可以删除旧索引。