第 17 章：数据写入与批量操作

这是《Elasticsearch 零基础实战指南》的独立章节版。本章从概念、实操和生产排查三个视角展开，代码块保留了原书可直接运行的版本。 Elasticsearch 写入不是简单调用一次 API。写入链路涉及协调节点、主分片、副本分片、refresh、translog、segment merge 和队列拒绝。设计不当的批量任务可以轻松拖垮一个集群。

本章覆盖写入链路、Bulk 策略、refresh 设置、并发限流、失败重试、同步链路和写入监控。

17.1 写入链路

客户端请求
-> coordinating node 解析路由
-> 转发主分片
-> 主分片写入 Lucene buffer 和 translog
-> 并行同步副本分片
-> 副本确认
-> 主分片返回协调节点
-> 协调节点返回客户端

写入成功通常表示主分片和满足策略的副本已接收，但文档可搜索还需要 refresh。

17.2 refresh 策略

查看设置：

GET /products/_settings/index.refresh_interval

修改：

PUT /products/_settings
{
  "index": {
    "refresh_interval": "30s"
  }
}

批量导入：

PUT /import-index/_settings
{
  "index": {
    "refresh_interval": "-1",
    "number_of_replicas": 0
  }
}

完成后：

POST /import-index/_refresh
PUT /import-index/_settings
{
  "index": {
    "refresh_interval": "1s",
    "number_of_replicas": 1
  }
}

导入期间副本为 0 有风险，必须确认目标集群和备份策略。

17.3 Bulk API 结构

POST /_bulk
{"index": {"_index": "products", "_id": "10001"}}
{"id": 10001, "title": "轻薄笔记本", "price": 6999}
{"create": {"_index": "products", "_id": "10002"}}
{"id": 10002, "title": "游戏手机", "price": 4999}
{"update": {"_index": "products", "_id": "10003"}}
{"doc": {"price": 499}}
{"delete": {"_index": "products", "_id": "10004"}}

要求：

1. 每行必须是完整 JSON；
2. 请求体必须是 NDJSON；
3. action 行和数据行成对出现；
4. delete 没有 data 行；
5. HTTP 200 不代表全部成功。

17.4 批次大小

常见批次：

不要盲目复制参数。文档大小从 1KB 到 100KB 时，同样的 5000 条批次压力完全不同。

17.5 写入队列与拒绝

常见错误：

{
  "type": "es_rejected_execution_exception",
  "reason": "rejected execution of coordinating operation"
}

含义：线程池队列满，当前节点来不及处理写入。

处理：

1. 降低客户端并发；
2. 减小批次；
3. 调大 refresh_interval；
4. 检查磁盘和 CPU；
5. 检查 Mapping 解析成本；
6. 检查副本数量；
7. 扩容数据节点；
8. 使用削峰队列。

17.6 Java Bulk 示例

public BulkResponse bulkProducts(List<Product> products) throws IOException {
    BulkRequest.Builder builder = new BulkRequest.Builder()
            .refresh(Refresh.False);

    for (Product product : products) {
        builder.operations(op -> op.index(index -> index
                .index("products")
                .id(String.valueOf(product.getId()))
                .document(product)
        ));
    }
    return client.bulk(builder.build());
}

失败处理：

BulkResponse response = client.bulk(request);
if (response.errors()) {
    List<FailedOperation> failed = response.items().stream()
            .filter(BulkItemResponse::isFailed)
            .map(item -> new FailedOperation(
                    item.index(),
                    item.id(),
                    item.error().reason()
            ))
            .toList();
    failedBuffer.addAll(failed);
}

17.7 自适应反压

固定并发容易在集群繁忙时失败。可以根据 429 和耗时调整并发：

public void adjustConcurrency(boolean rejected, long costMs) {
    if (rejected) {
        permits.release(Math.max(1, permits.availablePermits() / 2));
        return;
    }
    if (costMs > 1000) {
        return;
    }
    if (permits.availablePermits() < maxPermits) {
        permits.release();
    }
}

更简单可靠的方案：

1. 使用固定小并发压测；
2. 观察队列、CPU、IO、写入耗时；
3. 设置保守上限；
4. 429 指数退避；
5. 不在应用层无限重试。

17.8 数据同步模式

全量导入

适合：首次建索引、数据规模可控、可重复读取事实源

步骤：

1. 创建新索引；
2. 调整导入设置；
3. 分页读取数据库；
4. Bulk 写入；
5. 校验数量和抽样；
6. 恢复设置；
7. 切换别名。

增量同步

适合：数据有 updated_at 或版本号

SELECT *
FROM product
WHERE updated_at > ?
ORDER BY updated_at
LIMIT 1000;

注意：

1. 时间窗口要重叠，避免边界遗漏；
2. 相同 updated_at 的排序要稳定；
3. 使用上游版本做幂等；
4. 记录 checkpoint；
5. 处理乱序。

CDC 同步

MySQL -> binlog -> Kafka -> Connector -> Elasticsearch

优点：

1. 延迟低；
2. 可回放；
3. 覆盖删除；
4. 与应用解耦。

挑战：

1. schema 变更；
2. 乱序；
3. 重复事件；
4. 大事务；
5. 数据一致性校验。

17.9 幂等写入

使用业务 ID：

{"index": {"_index": "products", "_id": "10001"}}
{"id": 10001, "version": 128}

使用外部版本：

{"index": {"_index": "products", "_id": "10001", "version": 128, "version_type": "external"}}
{"id": 10001}

软删除：

{
  "id": 10001,
  "deleted": true,
  "deleted_at": "2026-08-25T10:00:00Z"
}

搜索时过滤：

{
  "term": { "deleted": false }
}

软删除保留数据便于恢复，但需要清理策略。

17.10 乱序与版本冲突

CDC 或消息重试可能乱序：

事件 v1
事件 v3
事件 v2

处理方式：

1. 使用 version_type=external 拒绝旧版本；
2. 按 ID 分区保证同实体顺序；
3. 使用 updated_at 比较后写入；
4. 对不可比较事件使用状态机；
5. 定期对账修正。

17.11 一致性校验

数量校验：

GET /products/_count

抽样校验：

GET /products/_search
{
  "size": 100,
  "query": {
    "range": {
      "updated_at": {
        "gte": "now-1h"
      }
    }
  }
}

对账维度：

17.12 写入监控指标

Prometheus 常用表达式示例：

rate(elasticsearch_indices_indexing_index_total[1m])
rate(elasticsearch_thread_pool_rejected_count{type="write"}[1m])
elasticsearch_indices_segment_count

17.13 写入上线清单

1. Mapping 已评审；
2. 批次大小已压测；
3. 并发上限明确；
4. 429 退避策略明确；
5. refresh_interval 合理；
6. translog 策略满足 RPO；
7. 同步任务可断点续跑；
8. 失败数据可重放；
9. 监控和告警已配置；
10. 已演练索引重建。

17.14 本章小结

• Bulk 是高吞吐写入的标准方式，但必须逐条处理失败；
• refresh_interval 影响吞吐和可见性延迟；
• 批次大小按字节数和压测结果确定；
• 写入拒绝需要反压，不要无限重试；
• 数据同步优先考虑幂等、断点、乱序和删除；
• CDC 常用 Kafka 作为可靠缓冲和回放层；
• 写入链路必须监控队列、拒绝、merge、refresh 和磁盘。

17.15 思考题

1. 为什么 HTTP 200 的 bulk 响应仍可能是失败？
2. 批量导入前将副本设为 0 有什么收益和风险？
3. 如何设计增量同步的 checkpoint？
4. CDC 事件乱序时如何避免旧数据覆盖新数据？
5. 写入出现大量 429 时应如何处理？