Elasticsearch 批量写入 Bulk 请求失败怎么查看具体错误信息？

Bulk 请求即使部分失败，HTTP 状态码通常也是 200，必须解析响应体中的 items 数组才能定位具体哪条数据出错。

核心观点：不要只看 HTTP 状态码，要检查响应 JSON 中的 errors 标志位和 items 数组里的具体状态码。

先确认响应体中 errors 字段是否为 true
先处理 items 数组中状态码非 200 或 201 的条目
再验证修正后的数据能否正常写入

实操示例：查看响应结构

在 Kibana Dev Tools 或 curl 中执行 Bulk 请求后，直接观察返回的 JSON 结构。重点关注顶层的 errors 字段和 items 数组。

以下示例模拟了一个字段类型冲突的场景：索引中 age 字段定义为 long 类型，但写入时传入了字符串。

POST _bulk
{ "index": { "_index": "test" } }
{ "name": "user1", "age": 25 }
{ "index": { "_index": "test" } }
{ "name": "user2", "age": "not_a_number" }

# 查看响应
{
  "errors": true, 
  "items": [
    { "index": { "status": 201 } },
    { 
      "index": { 
        "status": 400, 
        "error": { 
          "type": "mapper_parsing_exception",
          "reason": "failed to parse field [age] of type [long]" 
        } 
      } 
    }
  ]
}

机制解析

Elasticsearch 的 Bulk API 设计为“部分成功”模式。为了保证批量操作的整体吞吐量，即使其中某条文档写入失败，服务器也不会直接返回 500 或 400 错误中断整个请求，而是返回 200 OK，并在响应体中标记哪些子请求失败了。

这种机制意味着客户端必须主动解析响应内容，而不是依赖 HTTP 状态码来判断成功与否。如果不检查 items 数组，很容易误以为所有数据都已写入，导致数据丢失或不一致。

分步处理流程

1. 检查顶层 errors 标志

拿到响应后，首先查看顶层的 errors 字段。如果为 false，说明所有子请求都成功，无需进一步排查。如果为 true，说明至少有一条数据出错。

2. 遍历 items 数组定位错误

items 数组中的每个元素对应请求中的一条操作。遍历该数组，检查每个操作的状态码（status）。

200 或 201：表示成功。
400：通常是映射错误、字段类型不匹配或文档格式问题。
409：版本冲突，常见于更新操作。
429：写入队列已满，需要限流或扩容。
503：节点不可用或集群状态异常。

3. 提取错误详情

对于状态码异常的条目，查看其 error 字段。里面通常包含 type 和 reason，例如 mapper_parsing_exception 或 version_conflict_engine_exception。根据这些信息修正数据或配置。

4. 重试失败项

将失败的文档单独提取出来，修正后重新发送。不要直接重试整个 Bulk 包，以免重复写入成功的数据。

客户端代码处理示例

在实际开发中，通常使用客户端 SDK 进行批量写入。以下是 Python 和 Java 客户端解析错误响应的典型写法。

Python (elasticsearch-py):

from elasticsearch import Elasticsearch

es = Elasticsearch()
actions = [
    { "_index": "test", "name": "user1", "age": 25 },
    { "_index": "test", "name": "user2", "age": "not_a_number" }
]

response = es.bulk(operations=actions)

if response.get("errors"):
    for item in response.get("items", []):
        for op_type, op_result in item.items():
            if "error" in op_result:
                print(f"Error: {op_result['error']['reason']}")
                # 记录失败文档以便重试
                failed_doc = op_result.get("_source")

Java (RestHighLevelClient):

BulkRequest request = new BulkRequest();
// add actions...
BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);

if (response.hasFailures()) {
    for (BulkItemResponse item : response.getItems()) {
        if (item.isFailed()) {
            BulkItemResponse.Failure failure = item.getFailure();
            log.error("Failed doc id: {}, reason: {}", failure.getId(), failure.getMessage());
            // 处理失败项
        }
    }
}

典型错误场景复现

为了更好理解错误信息，以下是几种常见报错的触发场景及响应特征。

1. 字段类型不匹配 (400)

场景：索引映射中 age 为 long，写入字符串。

特征：error.type 为 mapper_parsing_exception。

2. 版本冲突 (409)

场景：使用 update 操作且指定 version，但当前文档版本已更新。

特征：error.type 为 version_conflict_engine_exception。

3. 写入拒绝 (429)

场景：并发过高，节点线程池队列满。

特征：error.type 为 es_rejected_execution_exception。

验证方法

1. 计数检查

使用 Count API 或 Kibana 界面查看索引文档总数，确认增加的数量与成功写入的数量一致。

GET /_cat/count/test?v

2. 查询特定文档

如果知道失败文档的 ID，尝试直接 GET 该文档，确认是否已存在或已更新。

3. 监控日志

如果是服务端错误（如 5xx），检查 Elasticsearch 集群日志，查看是否有 rejected execution 或 shard 异常。

常见注意事项

1. 忽略更新操作的版本控制

使用 update 操作时，如果外部版本管理不当，容易触发版本冲突错误。确保 version 或 if_seq_no 参数使用正确。

2. 单个文档过大

虽然 Bulk 请求有大小限制，但单个文档超过 http.max_content_length 也会导致解析失败。检查报错是否涉及 size 限制。

3. 映射严格模式

如果索引开启了 dynamic: strict，写入未定义字段会直接报错。检查映射配置是否允许新字段。

4. 客户端库封装差异

不同语言的客户端库对 Bulk 响应的封装不同。有的会抛出异常，有的需要手动检查响应对象。务必查阅所用客户端的官方文档。

参考来源

Elastic, Bulk API, https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html