如何使用 python 客户端库将 JSON 记录批量上传到 AWS OpenSearch 索引？

Question

我有一个足够大的数据集，我想在 AWS OpenSearch 中批量索引 JSON 对象。

我看不到如何使用以下任何一种来实现此目的：boto3、awswrangler、opensearch-py、elasticsearch、elasticsearch-py。

有没有办法在不直接使用 python 请求（PUT/POST）的情况下做到这一点？

请注意，这不适用于：ElasticSearch、AWS ElasticSearch。

非常感谢！

Answer 1

我终于找到了使用opensearch-py的方法，如下。

首先建立客户端，

# First fetch credentials from environment defaults
# If you can get this far you probably know how to tailor them
# For your particular situation. Otherwise SO is a safe bet :)
import boto3
credentials = boto3.Session().get_credentials()
region='eu-west-2' # for example
auth = AWSV4SignerAuth(credentials, region)

# Now set up the AWS 'Signer'
from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
auth = AWSV4SignerAuth(credentials, region)

# And finally the OpenSearch client
host=f"...{region}.es.amazonaws.com" # fill in your hostname (minus the https://) here
client = OpenSearch(
    hosts = [{'host': host, 'port': 443}],
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    connection_class = RequestsHttpConnection
)

呸！现在让我们创建数据：

# Spot the deliberate mistake(s) :D
document1 = {
    "title": "Moneyball",
    "director": "Bennett Miller",
    "year": "2011"
}

document2 = {
    "title": "Apollo 13",
    "director": "Richie Cunningham",
    "year": "1994"
}

data = [document1, document2]

提示！如果需要，请创建索引 -

my_index = 'my_index'

try:
    response = client.indices.create(my_index)
    print('\nCreating index:')
    print(response)
except Exception as e:
    # If, for example, my_index already exists, do not much!
    print(e)

这就是事情变得有点疯狂的地方。我没有意识到每个批量操作都需要一个，呃，action，例如“索引”、“搜索”等 - 现在让我们定义它

action={
    "index": {
        "_index": my_index
    }
}

下一个怪癖是 OpenSearch 批量 API 需要换行分隔的 JSON（请参阅https://www.ndjson.org），它基本上是将 JSON 序列化为字符串并用换行符分隔。有人在 SO 上写道，这个“奇怪”的 API 看起来像是由数据科学家设计的——我认为这远非冒犯。 （我同意 ndjson 很奇怪。）

可怕的是，现在让我们构建完整的 JSON 字符串，将数据和操作结合起来。一个助手 fn 就在眼前！

def payload_constructor(data,action):
    # "All my own work"

    action_string = json.dumps(action) + "\n"

    payload_string=""

    for datum in data:
        payload_string += action_string
        this_line = json.dumps(datum) + "\n"
        payload_string += this_line
    return payload_string

好的，现在我们终于可以调用批量 API。我想你可以混合各种动作（这里超出范围） - 去吧！

response=client.bulk(body=payload_constructor(data,action),index=my_index)

这可能是有史以来最无聊的妙语，但你有它。

您也可以直接获取 (geddit) .bulk() 以使用 index= 并将操作设置为：

action={"index": {}}

你好！

现在，选择你的毒药 - 其他解决方案看起来更短更整洁。

Answer 2

conn = wr.opensearch.connect(
         host=self.hosts, # URL
         port=443,
         username=self.username,
         password=self.password
    )

def insert_index_data(data, index_name='stocks', delete_index_data=False):
    """ Bulk Create 
        args: body [{doc1}{doc2}....]
    """
    if delete_index_data:
        index_name = 'symbol'
        self.delete_es_index(index_name)
    
    resp = wr.opensearch.index_documents(
         self.conn,
         documents=data,
         index=index_name   
     )
    print(resp)
    return resp

Answer 3

我使用下面的代码将 postgres 中的记录批量插入 OpenSearch (ES 7.2)

import sqlalchemy as sa
from sqlalchemy import text
import pandas as pd
import numpy as np
from opensearchpy import OpenSearch
from opensearchpy.helpers import bulk
import json

engine = sa.create_engine('postgresql+psycopg2://postgres:postgres@127.0.0.1:5432/postgres')

host = 'search-xxxxxxxxxx.us-east-1.es.amazonaws.com'
port = 443
auth = ('username', 'password') # For testing only. Don't store credentials in code.

# Create the client with SSL/TLS enabled, but hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True,
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    ssl_assert_hostname = False,
    ssl_show_warn = False
)

        


with engine.connect() as connection:
    result = connection.execute(text("select * from account_1_study_1.stg_pred where domain='LB'"))
    records = []
    for row in result:
        record = dict(row)
        record.update(record['item_dataset'])
        del record['item_dataset']
        records.append(record)
    df = pd.DataFrame(records)
    #df['Date'] = df['Date'].astype(str)
    df = df.fillna("null")
    print(df.keys)
    documents = df.to_dict(orient='records')

    #bulk(es ,documents, index='search-irl-poc-dump', raise_on_error=True)\
    
    #response=client.bulk(body=documents,index='sample-index')
    bulk(client, documents, index='search-irl-poc-dump', raise_on_error=True, refresh=True)