使用OpenSearch Python Bulk API将数据插入多个索引

Question

本文档显示了如何使用curl中的POST请求插入多个索引的批量数据：

如果我有这种格式的数据，则

[
{ "index": { "_index": "index-2022-06-08", "_id": "<id>" } }
{ "A JSON": "document" }
{ "index": { "_index": "index-2022-06-09", "_id": "<id>" } }
{ "A JSON": "document" }
{ "index": { "_index": "index-2022-06-10", "_id": "<id>" } }
{ "A JSON": "document" }
]

批量请求应从“ _ index”中获取索引名称：“ Index-2022-06-- 08“

我试图使用opensearch-py库进行相同的操作，但是我找不到任何示例代码段这样做。我正在使用这种格式发送AWS Lambda的请求。

client = OpenSearch(
            hosts = [{'host': host, 'port': 443}],
            http_auth = awsauth,
            use_ssl = True,
            verify_certs = True,
            connection_class = RequestsHttpConnection
            )
        
        resp = helpers.bulk(client, logs, index= index_name, max_retries = 3)

在这里，我已经提到index_name作为批量请求中的参数，因此它不会从数据本身中获取index_name。如果我不提及参数中的index_name，则会丢失错误4xx index_name。

我还在研究批量API源代码： https://github.com/opensearch-project/opensearch-py/blob/main/main/opensearchpy/helpers/actions.py#l373

看来Index_name似乎不是强制性参数。

谁能帮助我，我想念什么？

Answer 1

我遇到了同一问题，并在。当在 _Source-structure 中提供文档时，搜索 - 端点返回其工作。

批量方法的呼叫：

resp = helpers.bulk(
    self.opensearch,
    actions,
    max_retries=3,
)

其中操作是这样的字典列表：

[{
    '_op_type': 'update',
    '_index': 'index-name',
    '_id': 42,
    '_source': {
        "title": "Hello World!",
        "body": "..."
    }
}]

_OP_TYPE 可以用作定义操作的附加字段（ index ，<代码>更新， delete ，...）应为文档调用。

希望这有助于遇到同一问题的任何人！

Answer 2

使用以下代码希望您可以使用批量方法索引，有两种方法用于索引文档在OpenSearch中

from opensearchpy import OpenSearch, helpers
from opensearchpy.helpers import bulk

client = OpenSearch(
    hosts=[{"host": "localhost", "port": 9200}],
    http_auth=("admin", "admin"),
    use_ssl=True,
    verify_certs=False,
    ssl_assert_hostname=False,
    ssl_show_warn=False,
)

inputtobeindexed = [
    {"index": {"_index": "index-2022-06-08", "_id": "<id>"}},
    {"A JSON": "document"},
    {"index": {"_index": "index-2022-06-09", "_id": "<id>"}},
    {"A JSON": "document"},
    {"index": {"_index": "index-2022-06-10", "_id": "<id>"}},
    {"A JSON": "document"},
]

search_index_name = "yourindexname"
bulk_data = [
    {"_index": search_index_name, "_id": i, "_source": doc} for i, doc in enumerate()
]

bulk(client, bulk_data)