python - 在ElasticSearch中执行 "Transaction"

Question

假设我有一百万个对象需要保存到 ElasticSearch。保存对象的要求之一是只能在保存所有项目后才能搜索 - 否则结果(这与保存对象的不同属性的计数和总和有关 - 想想财务计算)将是错的。

这是我当前保存对象的代码:

from elasticsearch import Elasticsearch, helpers
ACTIONS = []
for item in HISTORY_DATA.values():
    ACTIONS.append({
        "_index": ES_INDEX_NAME,
        "_type": "_doc",
        "_id": item.pop('_id'),
        "_source": item
    })
_ = helpers.bulk(self.es, ACTIONS)

如何一次保存 20k 对象，但仅在保存所有项目后才“提交”事务？或者，如果我必须“一次保存所有内容”，我该怎么做？

Answer 1

Elasticsearch 没有事务，只有单个文档操作是原子的。

如果您的elasticsearch索引只有在插入所有项目后才可用，您可以使用此方法:

1. 将您的应用配置为使用索引别名。
   (此时别名不指向任何东西 - 没关系)
2. 创建索引(例如 index_1)并使用批量插入添加所有文档。
   (索引已准备好使用)
3. 将您的索引别名指向index_1。
   (您的应用程序可以使用index_1)
4. 如果您需要添加新一批项目，请创建新索引 index_2，将所有旧文档和新文档插入其中。执行任何所需的验证以确保一切正常。
   (对此索引的更改对您的应用程序来说是不可见的)
5. 将您的索引别名指向index_2。
   (这就像提交事务并切换到旧索引就像事务回滚)
6. 删除 index_1。

每当您需要添加新文档集时，请重复第 4、5、6 点。我在几个网站上使用这种方法，每个网站每次更新时都会将 10k 到 50k 文档添加到索引中。