python - 如何在Elasticsearch中批量插入而忽略过程中可能出现的所有错误？

Question

我使用的是 Elasticsearch 版本 6.8。

我需要将大约 10000 个文档(来自 csv 文件)插入到现有索引和映射索引中。

我正在使用python(版本3.7)代码:

    import csv  
    es = Elasticsearch();
    from elasticsearch import helpers
    with open(file_path) as f:
        reader = csv.DictReader(f)
        helpers.bulk(es, reader, index=index_name, doc_type=doc_type)

但是我收到错误:

raise BulkIndexError("%i document(s) failed to index." % len(errors), errors)
elasticsearch.helpers.errors.BulkIndexError: ('3 document(s) failed to index.'

出现此错误的原因是 csv 文件中的某些值是字符串值而不是浮点值。

499 个文档后批量停止，应用程序崩溃。

有没有办法批量处理所有文档(~10000)，如果出现错误(由于映射或错误的值)，请告诉 python/elastic忽略那些文档并继续批量操作？

Answer 1

您可以将参数 raise_on_error 设置为 False，因为默认情况下它是 True，如 Python bulk documentation 中的建议。 。它应该看起来像这样:

helpers.bulk(es, reader, index=index_name, doc_type=doc_type, raise_on_error=False)

请记住:

When errors are being collected original document data is included in the error dictionary which can lead to an extra high memory usage. If you need to process a lot of data and want to ignore/collect errors please consider using the streaming_bulk() helper which will just return the errors and not store them in memory.

您还可以查看此 Python ES bulk Examples 中的示例 12、25 和 39