如何使用Elasticsearch dsl py使Elasticsearch索引保持最新?

我开发了一个小的个人信息目录,我的客户可以通过Django管理界面访问和更新该目录。这些信息需要可搜索,所以我建立了Django网站,将这些数据保存在搜索索引中。我最初使用Haystack和Whoosh作为搜索索引,但最近我不得不放弃这些工具,转而使用Elasticsearch 5。

以前,每当更新目录中的任何内容时,代码只需清除整个搜索索引,然后从头开始重新构建它。这个目录中只有几百个条目,所以这并不是严重的不合格。不幸的是,在Elasticsearch中尝试做同样的事情是非常不可靠的,因为我假设在我的代码中存在某种竞争条件。

以下是我编写的使用elasticsearch py和elasticsearch dsl py的代码:

import elasticsearch
import time
from django.apps import apps
from django.conf import settings
from elasticsearch.helpers import bulk
from elasticsearch_dsl.connections import connections
from elasticsearch_dsl import DocType, Text, Search

# Create the default Elasticsearch connection using the host specified in settings.py.
elasticsearch_host = "{0}:{1}".format(
    settings.ELASTICSEARCH_HOST['HOST'], settings.ELASTICSEARCH_HOST['PORT']
)
elasticsearch_connection = connections.create_connection(hosts=[elasticsearch_host])


class DepartmentIndex(DocType):
    url = Text()
    name = Text()
    text = Text(analyzer='english')
    content_type = Text()

    class Meta:
        index = 'departmental_directory'


def refresh_index():
    # Erase the existing index.
    try:
        elasticsearch_connection.indices.delete(index=DepartmentIndex().meta.index)
    except elasticsearch.exceptions.NotFoundError:
        # If it doesn't exist, the job's already done.
        pass

    # Wait a few seconds to give enough time for Elasticsearch to accept that the 
    # DepartmentIndex is gone before we try to recreate it.
    time.sleep(3)

    # Rebuild the index from scratch.
    DepartmentIndex.init()
    Department = apps.get_model('departmental_directory', 'Department')
    bulk(
        client=elasticsearch_connection, 
        actions=(b.indexing() for b in Department.objects.all().iterator())
    )

我已经设置了Django的呼叫信号 refresh_index() 每当 Department 已保存。但是 refresh\u index() 由于以下错误经常崩溃:

elasticsearch.exceptions.RequestError: TransportError(400, u'index_already_exists_exception', u'index [departmental_directory/uOQdBukEQBWvMZk83eByug] already exists')

这就是为什么我加上 time.sleep(3) 呼叫我假设索引到那时还没有被完全删除 DepartmentIndex.init() 调用,这是导致错误的原因。

我的猜测是,我只是用了一种完全错误的方式。 必须有更好的方法使用elasticsearch dsl py来保持elasticsearch索引的最新状态,但我不知道它是什么,而且我还无法通过他们的文档找到它。

在google上搜索“从头开始重建elasticsearch索引”会得到大量关于“如何重新索引elasticsearch数据”的结果,但这不是我想要的。我需要 代替 这些数据包含来自我的应用程序数据库的新的、更为最新的数据。