使用 Elasticsearch _bulk API 命令高效管理数据

Elasticsearch 是一个强大的分布式搜索和分析引擎，以其速度和可伸缩性而闻名。随着数据量的增长和应用需求的增加，优化与集群的交互方式变得至关重要。提高性能（尤其是数据摄取和修改）最有效的方法之一是利用 _bulk API。此命令允许您将多个索引、更新和删除操作合并到一个高效的请求中，从而显著减少网络开销并提高整体吞吐量。

本文将指导您了解 _bulk API 的结构，并演示如何使用它的实际示例来简化在 Elasticsearch 中的数据管理操作。通过掌握 _bulk API，您可以释放巨大的性能提升，并使您的 Elasticsearch 交互更加高效。

理解 _bulk API 结构

The _bulk API 通过接受一系列操作及其相关的元数据和数据来工作。每个操作都在单独的一行上定义，并且这些行由换行符 (\n) 分隔。请求正文本质上是一系列 JSON 对象，每个对象代表一个操作。该 API 期望这些操作具有特定的格式，通常涉及一个“操作和元数据”行，后跟一个包含文档数据的“源”行。

_bulk 请求的关键组成部分：

操作和元数据行： 此行指定操作类型（例如 index、create、update、delete）、目标索引，以及可选的文档类型和 ID。对于 index 和 create 操作，文档 ID 是可选的；如果省略，Elasticsearch 将自动生成一个。
源行： 此行包含要索引或更新的实际 JSON 文档。对于 delete 操作，此行将被省略。
换行符分隔符： 每个操作/元数据对及其对应的源（如果适用）必须由换行符 (\n) 分隔。整个请求正文应以换行符 (\n) 结尾。

示例结构：

{ "action_and_metadata_line" }
{ "source_line" }
{ "action_and_metadata_line" }
{ "source_line" }
...

或者对于删除操作：

{ "action_and_metadata_line" }
...

使用 _bulk 执行常见操作

The _bulk API 非常通用，可以在单个请求中处理混合操作。这正是其真正的强大之处，允许您在一次往返中执行复杂的数据操作。

索引多个文档

要索引多个文档，您可以使用 index 操作。如果具有指定 ID 的文档已存在，index 将覆盖它。如果您希望确保仅在文档尚不存在时才对其进行索引，请改用 create 操作。

示例： 索引两个新文档。

POST /_bulk
{
  "index": { "_index": "my-index", "_id": "1" }
}
{
  "field1": "value1",
  "field2": "value2"
}
{
  "index": { "_index": "my-index", "_id": "2" }
}
{
  "field1": "another_value",
  "field2": "different_value"
}

更新文档

可以使用 update 操作来更新文档。您需要指定要更新的文档 ID，并提供一个包含要更改的字段的部分文档。如果您想使用脚本进行更新，可以在 update 操作内部执行此操作。

示例： 更新现有文档中的一个字段。

POST /_bulk
{
  "update": { "_index": "my-index", "_id": "1" }
}
{
  "doc": {
    "field1": "updated_value"
  }
}

删除文档

要删除文档，请使用 delete 操作，指定要删除的文档的 _index 和 _id。删除操作不需要源文档。

示例： 删除一个文档。

POST /_bulk
{
  "delete": { "_index": "my-index", "_id": "2" }
}

合并操作

真正的效率来自于混合这些操作。您可以在同一个 _bulk 请求中索引新文档、更新现有文档以及删除其他文档。

示例： 在一个请求中进行索引、更新和删除。

POST /_bulk
{
  "index": { "_index": "my-index", "_id": "3" }
}
{
  "field1": "new_document_field",
  "field2": "new_document_value"
}
{
  "update": { "_index": "my-index", "_id": "1" }
}
{
  "doc": {
    "field1": "further_updated_value"
  }
}
{
  "delete": { "_index": "my-index", "_id": "2" }
}

响应处理

The _bulk API 返回一个 JSON 响应，其中详细说明了每个单独操作的结果。解析此响应以验证所有操作是否成功以及识别任何错误都至关重要。

响应将包含一个 items 数组，其中每个元素对应您请求中的一个操作，顺序相同。每个项目将包括 index、create、update 或 delete 操作，以及其状态（例如 created、updated、deleted、noop）和其他相关元数据。

示例响应片段：

{
  "took": 150,
  "errors": false,
  "items": [
    {
      "index": {
        "_index": "my-index",
        "_id": "3",
        "version": 1,
        "result": "created",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 0,
        "_primary_term": 1
      }
    },
    {
      "update": {
        "_index": "my-index",
        "_id": "1",
        "version": 2,
        "result": "updated",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 1,
        "_primary_term": 1
      }
    },
    {
      "delete": {
        "_index": "my-index",
        "_id": "2",
        "version": 2,
        "result": "deleted",
        "_shards": {"total": 2, "successful": 1, "failed": 0},
        "_seq_no": 2,
        "_primary_term": 1
      }
    }
  ]
}

如果任何操作失败，响应中的顶级 errors 字段将为 true，并且失败操作的单个项目将包含一个详细说明问题的 error 对象。

最佳实践和技巧

批次大小： 尽管 _bulk API 很高效，但超大的批次仍然会给资源带来压力。试验以找到适合您的集群和用例的最佳批次大小。一个常见的起点是每批 1,000 到 5,000 个文档。
错误处理： 始终解析响应以查找错误。如有必要，为瞬态错误实现重试逻辑。
换行符分隔符： 确保在每个 JSON 对象之间正确使用换行符 (\n)。格式不正确是 _bulk API 失败的常见原因。
并行化： 对于非常高的摄取率，考虑并行发送多个 _bulk 请求，但要注意集群的容量。
create 与 index： 当您想避免意外覆盖现有文档时，请使用 create。对于通用的 upsert（更新或插入）行为，请使用 index。
API 客户端： 大多数 Elasticsearch 客户端库都提供方便的方法来构建和执行 _bulk 请求，从而抽象掉一些手动格式设置。

结论

对于任何希望优化数据操作的人来说，Elasticsearch _bulk API 都是不可或缺的工具。通过将多个索引、更新和删除请求整合到一个 API 调用中，您可以显著减少网络延迟，提高处理效率，并增强 Elasticsearch 集群的整体性能。了解其结构并有效地实施它将带来更强大、更可扩展的数据管理策略。