Elasticsearch

本文集文档发布于灏天文库

4.5 批量操作 (Bulk API)

文档摘要

4.5 批量操作 (Bulk API) 4.5 Elasticsearch 批量操作 (Bulk API) Elasticsearch 的 Bulk API 允许你在单个请求中执行多个操作，例如索引、更新、删除文档。这对于提升性能至关重要，因为它减少了客户端和 Elasticsearch 集群之间的往返次数。 4.5.1 Bulk API 的优势 性能提升： 通过减少网络开销，显著提高数据处理速度。 原子性： 虽然 Bulk 请求本身不是原子性的，但每个操作都是独立执行的。如果一个操作失败，不会影响其他操作。 简化代码： 将多个操作组合成一个请求，简化了客户端代码。 4.5.

4.5 批量操作 (Bulk API)

4.5 Elasticsearch 批量操作 (Bulk API)

Elasticsearch 的 Bulk API 允许你在单个请求中执行多个操作，例如索引、更新、删除文档。这对于提升性能至关重要，因为它减少了客户端和 Elasticsearch 集群之间的往返次数。

4.5.1 Bulk API 的优势

  • 性能提升： 通过减少网络开销，显著提高数据处理速度。

  • 原子性： 虽然 Bulk 请求本身不是原子性的，但每个操作都是独立执行的。如果一个操作失败，不会影响其他操作。

  • 简化代码： 将多个操作组合成一个请求，简化了客户端代码。

4.5.2 Bulk API 的基本结构

Bulk API 使用一种特殊的 JSON 格式，它由两部分组成：

  1. 元数据行： 指定要执行的操作（index、create、update、delete）和相关的元数据（索引、类型、ID）。

  2. 数据行（可选）： 包含要索引或更新的文档数据。

{ "action": { "metadata" } }
{ "data" }
{ "action": { "metadata" } }
{ "data" }
...
​

其中，action 可以是以下之一：

  • index: 索引一个文档。如果文档已存在，则替换它。

  • create: 索引一个文档。如果文档已存在，则操作失败。

  • update: 更新一个文档。需要指定 doc 或 script。

  • delete: 删除一个文档。

metadata 包含关于操作的信息，例如 _index、_id、_type (在 Elasticsearch 中已弃用，8.x 中已移除)。

data 包含要索引或更新的实际文档数据（JSON 格式）。

4.5.3 Bulk API 的代码实践

以下是一些使用 Bulk API 的代码示例（使用 Python 和 Elasticsearch 官方客户端）。

1. 索引多个文档

from elasticsearch import Elasticsearch
# 连接到 Elasticsearch 集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 准备 Bulk 数据
bulk_data = [
    {'index': {'_index': 'my_index', '_id': 1}},
    {'title': 'Document 1', 'content': 'This is the first document.'},
    {'index': {'_index': 'my_index', '_id': 2}},
    {'title': 'Document 2', 'content': 'This is the second document.'},
    {'index': {'_index': 'my_index', '_id': 3}},
    {'title': 'Document 3', 'content': 'This is the third document.'}
]
# 将 Bulk 数据转换为字符串
bulk_str = '\n'.join(map(lambda x: json.dumps(x), bulk_data)) + '\n'
# 执行 Bulk 请求
response = es.bulk(body=bulk_str)
# 打印响应
print(response)
​

解释:

  • 首先，我们连接到 Elasticsearch 集群。

  • 然后，我们准备一个包含多个 index 操作的 bulk_data 列表。 每个 index 操作都包含元数据（_index 和 _id）以及要索引的文档数据。

  • 我们将 bulk_data 列表转换为一个字符串，其中每个操作和数据都用换行符分隔。 这是 Bulk API 所需的格式。

  • 最后，我们使用 es.bulk() 方法执行 Bulk 请求，并将 bulk_str 作为请求体传递。

  • 打印响应，查看操作结果。

2. 创建多个文档 (如果不存在)

from elasticsearch import Elasticsearch
import json
# 连接到 Elasticsearch 集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 准备 Bulk 数据
bulk_data = [
    {'create': {'_index': 'my_index', '_id': 4}},
    {'title': 'Document 4', 'content': 'This is the fourth document.'},
    {'create': {'_index': 'my_index', '_id': 5}},
    {'title': 'Document 5', 'content': 'This is the fifth document.'}
]
# 将 Bulk 数据转换为字符串
bulk_str = '\n'.join(map(lambda x: json.dumps(x), bulk_data)) + '\n'
# 执行 Bulk 请求
response = es.bulk(body=bulk_str)
# 打印响应
print(response)
​

解释:

这个例子与前面的例子类似，但使用了 create 操作而不是 index 操作。 如果具有指定 ID 的文档已经存在，create 操作将失败。

3. 更新多个文档

from elasticsearch import Elasticsearch
import json
# 连接到 Elasticsearch 集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 准备 Bulk 数据
bulk_data = [
    {'update': {'_index': 'my_index', '_id': 1}},
    {'doc': {'content': 'This is the updated first document.'}},
    {'update': {'_index': 'my_index', '_id': 2}},
    {'doc': {'content': 'This is the updated second document.'}}
]
# 将 Bulk 数据转换为字符串
bulk_str = '\n'.join(map(lambda x: json.dumps(x), bulk_data)) + '\n'
# 执行 Bulk 请求
response = es.bulk(body=bulk_str)
# 打印响应
print(response)
​

解释:

  • 这里使用 update 操作来更新现有文档。

  • doc 键指定要更新的字段和值。

  • 如果文档不存在，更新操作将失败。

4. 删除多个文档

from elasticsearch import Elasticsearch
import json
# 连接到 Elasticsearch 集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 准备 Bulk 数据
bulk_data = [
    {'delete': {'_index': 'my_index', '_id': 3}},
    {'delete': {'_index': 'my_index', '_id': 4}}
]
# 将 Bulk 数据转换为字符串
bulk_str = '\n'.join(map(lambda x: json.dumps(x), bulk_data)) + '\n'
# 执行 Bulk 请求
response = es.bulk(body=bulk_str)
# 打印响应
print(response)
​

解释:

  • 这里使用 delete 操作来删除文档。

  • 只需要指定文档的 _index 和 _id。

5. 混合操作

Bulk API 允许你在同一个请求中混合不同的操作。

from elasticsearch import Elasticsearch
import json
# 连接到 Elasticsearch 集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 准备 Bulk 数据
bulk_data = [
    {'index': {'_index': 'my_index', '_id': 6}},
    {'title': 'Document 6', 'content': 'This is the sixth document.'},
    {'update': {'_index': 'my_index', '_id': 1}},
    {'doc': {'content': 'This is the updated and indexed first document.'}},
    {'delete': {'_index': 'my_index', '_id': 2}}
]
# 将 Bulk 数据转换为字符串
bulk_str = '\n'.join(map(lambda x: json.dumps(x), bulk_data)) + '\n'
# 执行 Bulk 请求
response = es.bulk(body=bulk_str)
# 打印响应
print(response)
​

解释:

这个例子展示了如何在同一个 Bulk 请求中执行 index、update 和 delete 操作。

4.5.4 Bulk API 的响应

Bulk API 的响应是一个 JSON 对象，包含有关每个操作结果的信息。

{
  "took": 123,
  "errors": false,
  "items": [
    {
      "index": {
        "_index": "my_index",
        "_id": "1",
        "_version": 1,
        "result": "created",
        "_shards": {
          "total": 2,
          "successful": 1,
          "failed": 0
        },
        "status": 201,
        "_seq_no": 0,
        "_primary_term": 1
      }
    },
    {
      "update": {
        "_index": "my_index",
        "_id": "2",
        "_version": 2,
        "result": "updated",
        "_shards": {
          "total": 2,
          "successful": 1,
          "failed": 0
        },
        "status": 200,
        "_seq_no": 1,
        "_primary_term": 1
      }
    },
    {
      "delete": {
        "_index": "my_index",
        "_id": "3",
        "_version": 1,
        "result": "deleted",
        "_shards": {
          "total": 2,
          "successful": 1,
          "failed": 0
        },
        "status": 200,
        "_seq_no": 2,
        "_primary_term": 1
      }
    }
  ]
}
​

  • took: 请求花费的时间（毫秒）。

  • errors: 一个布尔值，指示是否发生任何错误。 如果为 true，则需要检查 items 数组以查找错误。

  • items: 一个数组，包含每个操作的结果。 每个结果都包含有关操作的元数据（_index、_id、_version 等）和状态代码。

4.5.5 Bulk API 的最佳实践

  • 批量大小： 没有一个适用于所有情况的最佳批量大小。 需要根据你的数据大小、硬件和网络条件进行实验。 通常，几百到几千个操作的批量大小是一个好的起点。

  • 错误处理： 始终检查 Bulk API 的响应以查找错误。 如果发生错误，请记录错误并采取适当的措施（例如，重试失败的操作）。

  • 数据格式： 确保 Bulk 数据的格式正确。 错误的格式会导致 Bulk 请求失败。

  • 性能监控： 监控 Elasticsearch 集群的性能，以确保 Bulk API 没有导致任何性能问题。

4.5.6 Bulk API 的流程图

流程图解释:

  1. 开始: 流程的起点。

  2. 准备 Bulk 数据: 创建包含多个操作的列表或数据结构。

  3. 转换为 Bulk 字符串: 将准备好的数据转换为 Bulk API 所需的字符串格式。

  4. 执行 Bulk 请求: 将 Bulk 字符串发送到 Elasticsearch 集群。

  5. 检查响应: 分析 Elasticsearch 返回的响应，以确定是否发生任何错误。

  6. errors=true: 如果响应中指示存在错误，则执行错误处理流程。

  7. 处理错误: 包含以下步骤：

    • 记录错误信息: 将错误信息记录到日志或其他监控系统中。

    • 重试 (可选): 可以选择重试失败的操作。

  8. errors=false: 如果响应中没有指示错误，则流程成功完成。

  9. 完成: Bulk 操作已成功执行。

  10. 结束: 流程的终点。

4.5.7 总结

Elasticsearch 的 Bulk API 是一个强大的工具，可以显著提高数据处理性能。 通过理解 Bulk API 的基本结构、代码实践和最佳实践，你可以有效地利用它来优化你的 Elasticsearch 应用程序。 记住要始终检查 Bulk API 的响应以查找错误，并根据你的数据大小和硬件条件调整批量大小。

发布者: 作者: 转发
评论区 (0)
U
0/500