如何使用Python客户端库向AWS OpenSearch索引批量上传JSON记录

不需要手动调用requests构造PUT/POST请求，直接用官方封装的SDK即可实现，以下两种方案均适配AWS OpenSearch（不涉及旧版ElasticSearch、AWS ElasticSearch场景）：

方案1：使用官方opensearch-py SDK的批量助手 #

这是OpenSearch官方维护的Python客户端，所有HTTP请求逻辑、AWS V4签名、重试机制都已内置，你只需要传入待导入的JSON数据即可。

• 安装依赖

  pip install opensearch-py requests-aws4auth boto3
  

• 初始化客户端

  from opensearchpy import OpenSearch, RequestsHttpConnection
  from requests_aws4auth import AWS4Auth
  import boto3
  
  # 自动读取本地AWS凭证，支持环境变量、~/.aws/credentials、IAM角色等多种凭证方式
  creds = boto3.Session().get_credentials()
  aws_auth = AWS4Auth(
      region="你的OpenSearch集群所在区域",
      service="es",
      refreshable_credentials=creds
  )
  
  client = OpenSearch(
      hosts=[{"host": "你的OpenSearch访问域名", "port": 443}],
      http_auth=aws_auth,
      use_ssl=True,
      verify_certs=True,
      connection_class=RequestsHttpConnection,
      pool_maxsize=30
  )
  

• 执行批量导入
  用SDK内置的bulk助手支持生成器逐批读取数据，大数据集不需要全量加载到内存：

  from opensearchpy.helpers import bulk
  
  # 数据生成器，逐行产出待导入文档，适配TB级大数据集
  def doc_generator(dataset_iter, target_index):
      for idx, doc in enumerate(dataset_iter):
          yield {
              "_index": target_index,
              "_id": idx, # 可自定义文档ID规则，不传则自动生成
              "_source": doc
          }
  
  success_cnt, failed_items = bulk(
      client=client,
      actions=doc_generator(你的JSON数据集迭代器, "目标索引名"),
      chunk_size=500, # 单批提交文档数，单条文档大就调小，反之调大，范围200-2000即可
      max_retries=3,
      raise_on_error=False,
      request_timeout=60
  )
  

注意：不要用旧版elasticsearch-py客户端连接OpenSearch，两个项目分叉后存在接口兼容性问题，直接使用官方维护的opensearch-py即可。

方案2：使用awswrangler的封装接口（适合结构化数据集） #

如果你的数据集是pandas DataFrame、JSON列表/迭代器格式，可以直接用awswrangler封装好的OpenSearch写入方法，连客户端初始化逻辑都不需要手动写：

• 安装依赖

  pip install awswrangler pandas
  

• 批量写入示例

  import awswrangler as wr
  import pandas as pd
  
  # DataFrame格式数据集一行写入
  wr.opensearch.index_df(
      df=pd.DataFrame(你的数据集),
      index="目标索引名",
      endpoint="https://你的OpenSearch访问域名",
      aws_region="你的OpenSearch集群所在区域",
      bulk_size=500
  )
  
  # JSON格式迭代器/列表写入
  wr.opensearch.index_documents(
      documents=你的JSON数据集迭代器,
      index="目标索引名",
      endpoint="https://你的OpenSearch访问域名",
      aws_region="你的OpenSearch集群所在区域",
      bulk_size=500
  )
  

该方法内部已经处理了内存优化、批量分片、签名认证、错误重试，大文件导入不会出现OOM问题。

以上两种方案的所有HTTP请求逻辑都由依赖库内部封装，全程不需要你手动调用requests构造PUT/POST请求。之前查文档没找到对应实现，大概率是搜索时混淆了OpenSearch和旧版ElasticSearch的关键词，注意OpenSearch独立迭代后客户端已经和ES客户端拆分维护，不要混用。

内容的提问来源于stack exchange，提问作者jtlz2