一日一技：Elasticsearch批量插入时，存在就不插入
摄影：产品经理
买单：kingname
当我们使用 Elasticsearch-py 批量插入数据到 ES 的时候，我们常常使用它的 helpers模块里面的bulk函数。其使用方法如下：
from elasticsearch import helpers, Elasticsearch
es = Elasticsearch(xxx)
def generator():
datas = [1, 2, 3]
for data in datas:
yield {
'_id': "xxx",
'_source': {
'age': data
}
}
helpers.bulk(es,
index='xxx',
generator(),
doc_type='doc',)
但这种方式有一个问题，它默认相当于upsert操作。如果_id 对应的文档已经在 ES 里面了，那么数据会被更新。如果_id 对应的文档不在 ES 中，那么就插入。
如果我想实现，不存在就插入，存在就跳过怎么办？此时就需要在文档里面添加_op_type指定操作类型为create:
from elasticsearch import helpers, Elasticsearch
es = Elasticsearch(xxx)
def generator():
datas = [1, 2, 3]
for data in datas:
yield {
'_op_type': 'create',
'_id': "xxx",
'_source': {
'age': data
}
}
helpers.bulk(es,
generator(),
index='xxx',
doc_type='doc')
此时，如果_id 对应的文档不在 ES 中，那么就会正常插入，如果ES里面已经有_id对应的数据了，那么就会报错。由于bulk一次性默认插入500条数据，假设其中有2条数据已经存在了，那么剩下的498条会被正常插入。然后程序报错退出，告诉你有两条写入失败，因为已经存在。
如果你不想让程序报错终止，那么可以增加2个参数：
helpers.bulk(es,
generator(),
index='xxx',
doc_type='doc',
raise_on_exception=False, raise_on_error=False)
其中raise_on_exception=False表示在插入数据失败时，不需要抛出异常。raise_on_error=False表示不抛出BulkIndexError。
本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2019-09-25，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
相关产品与服务
Elasticsearch Service
腾讯云 Elasticsearch Service（ES）是云端全托管海量数据检索分析服务，拥有高性能自研内核，集成X-Pack。ES 支持通过自治索引、存算分离、集群巡检等特性轻松管理集群，也支持免运维、自动弹性、按需使用的 Serverless 模式。使用 ES 您可以高效构建信息检索、日志分析、运维监控等服务，它独特的向量检索还可助您构建基于语义、图像的AI深度应用。
腾讯云开发者
