使用streaming_bulk()时遇409错误，确认文档仅上传一次的问题

针对你遇到的700万带嵌套结构文档批量上传时，使用streaming_bulk出现409错误（已确认每份文档仅被包含一次）的问题，结合你的拆分上传场景（260个文件，每个约3万条action），我给你梳理下可能的原因和对应的解决方案：

先排查核心原因 #

1. 有没有隐性的重复ID？ #

你说每份文档只包含一次，但要警惕不同文件里出现了相同的_id——比如生成_id时，是不是每个文件的自增计数器都从头开始了？或者生成逻辑有漏洞导致全局ID重复。这是409最常见的原因，哪怕你主观觉得没有，也一定要用工具校验。

2. ES集群的状态延迟 #

如果是分布式ES集群，分片的元数据同步可能有延迟：比如前一批次的create操作还没完全落盘、同步到所有节点，后续批次的请求就发过来了，集群可能误判为重复提交。

3. 重试机制导致的重复提交 #

streaming_bulk默认有重试逻辑，如果之前的请求因为超时、网络波动等临时错误失败，重试时会重新发送同一个action，这也可能触发409。

针对性解决方案 #

1. 先做全局ID唯一性校验 #

写个简单的脚本遍历所有260个文件，统计每个_id的出现次数，彻底排除重复问题：

from collections import defaultdict
import json
import glob

# 替换成你的文件路径匹配规则，比如"./actions_files/*.json"
file_list = glob.glob("./actions_files/*.json")
id_counter = defaultdict(int)

for file_path in file_list:
    with open(file_path, 'r', encoding='utf-8') as f:
        try:
            actions = json.load(f)
            for action in actions:
                doc_id = action['_id']
                id_counter[doc_id] += 1
                if id_counter[doc_id] > 1:
                    print(f"⚠️ 发现重复ID {doc_id}，出现在文件 {file_path}")
                    exit(1)
        except Exception as e:
            print(f"读取文件 {file_path} 出错: {str(e)}")

print("✅ 所有文档ID全局唯一")

2. 调整streaming_bulk的重试与错误处理 #

关闭不必要的重试，同时显式处理409错误，避免无效的重复提交：

from elasticsearch import Elasticsearch
from elasticsearch.helpers import streaming_bulk

# 初始化ES客户端，替换成你的集群配置
es = Elasticsearch(["http://your-es-host:9200"], basic_auth=("user", "pass"))

def action_generator():
    # 遍历所有文件生成action
    for file_path in glob.glob("./actions_files/*.json"):
        with open(file_path, 'r', encoding='utf-8') as f:
            actions = json.load(f)
            yield from actions

# 执行批量上传，关闭重试，同时捕获409错误
for success, result in streaming_bulk(
    client=es,
    actions=action_generator(),
    chunk_size=1000,  # 调小chunk size，降低单次请求压力
    max_retries=0,    # 关闭重试，避免重复提交
    raise_on_error=False,
    raise_on_exception=False
):
    action_details = result['create']
    if not success:
        if action_details['status'] == 409:
            print(f"📝 文档ID {action_details['_id']} 已存在，自动跳过")
        else:
            print(f"❌ 处理失败，ID: {action_details['_id']}，错误信息: {action_details.get('error')}")

3. 可选：改用op_type='index'兜底 #

如果你的业务逻辑允许覆盖已存在的文档（哪怕理论上不会重复），可以把action里的_op_type从create改成index，这样即使碰到重复ID也会直接更新，不会返回409错误：

# 修改后的action格式
esActionFromFile = [{ 
    '_index': 'mt-interval-test-9', 
    '_type': 'doc', 
    '_id': 5641254, 
    '_source': a, 
    '_op_type': 'index'  # 替换create为index
}]

4. 检查ES集群健康状态 #

先确保集群状态是green，分片没有未分配的情况，避免因为集群异常导致的误判：

curl -X GET "http://your-es-host:9200/_cluster/health?pretty"

如果是yellow状态，先解决分片分配问题再继续上传。

额外优化建议 #

• 对于700万级别的批量上传，parallel_bulk的效率通常比streaming_bulk更高，可以设置thread_count为CPU核心数的2倍左右，同时搭配合适的chunk_size（比如1000-5000），平衡并发和内存占用。
• 你当前每个文件3万条的规模有点大，可以拆成1万条/文件，减少单次读取文件的内存压力，也降低批量请求的失败概率。

内容的提问来源于stack exchange，提问作者labourday