Python操作Elasticsearch批量更新时如何仅推送新增文档

方案1：基于文档ID的幂等写入（最便捷，无需额外存储） #

利用ES原生create操作的幂等特性：只有对应_id的文档不存在时才会写入，已存在则直接返回冲突错误，不会覆盖或重复写入数据，适合数据量不大的场景。

修改后代码示例 #

1. 调整生成批量动作的自定义函数：

def my_function(df):
    for c, line in enumerate(df):
        yield {
            '_index': 'my_index',
            # ES7及以上版本无需指定_type，默认就是_doc可直接省略
            '_type': '_doc',
            '_id': line.get("_id", None),
            # 新增op_type配置，仅允许创建不存在的文档
            'op_type': 'create',
            '_source': {
                'field_A': line.get('field', "")
            }
        }
    # 注意：Python 3.7+ 生成器迭代结束会自动抛出StopIteration，手动触发会报警告，建议删除此行

2. 调整bulk调用参数，忽略重复文档的冲突错误不中断执行：

try:
    # 新增参数：遇到单条文档错误不抛出全局异常，继续执行后续写入
    res = helpers.bulk(es, my_function(df), raise_on_error=False, raise_on_exception=False)
    print(f"推送完成，成功写入{res[0]}条，失败{len(res[1])}条（失败多为已存在的重复文档）")
except Exception as e:
    print(e)

方案2：基于增量标记的前置过滤（适合大数据量，性能更高） #

如果源数据有明确的增量标识字段（比如自增ID、创建时间、更新时间），可以在推送前就过滤掉已处理过的数据，减少不必要的ES交互，性能远高于方案1。

实现逻辑 #

• 每次脚本执行完成后，把本次处理的最大增量标识（比如最大的创建时间、最大的自增ID）存储到本地文件、Redis或者ES专门的管理索引中
• 下次脚本启动时先读取上次存储的增量标识，仅从数据源中筛选大于该标识的新增数据生成df，再执行推送逻辑，可配合方案1的op_type做双重校验

示例代码 #

import os
# 读取上次处理的最大时间戳
last_process_time = "1970-01-01 00:00:00"
if os.path.exists("last_process.log"):
    with open("last_process.log", "r") as f:
        last_process_time = f.read().strip()

# 仅筛选新增数据，假设df存在create_time作为增量标识
new_df = df[df["create_time"] > last_process_time]

try:
    res = helpers.bulk(es, my_function(new_df), raise_on_error=False, raise_on_exception=False)
    # 处理完成更新最大时间戳到本地文件
    max_time = new_df["create_time"].max()
    with open("last_process.log", "w") as f:
        f.write(str(max_time))
    print(f"新增数据推送完成，成功写入{res[0]}条")
except Exception as e:
    print(e)

优化建议 #

• 如果不需要兼容ES6及更早版本，可以去掉代码中的_type字段，ES7开始已经废弃自定义类型，默认使用_doc无需显式声明
• 可以解析返回的res[1]错误信息判断失败原因：返回409状态码为重复文档，其他错误（比如字段类型不匹配）可单独打印日志排查

内容的提问来源于stack exchange，提问作者sam.marhaendra