使用awswrangler向AWS OpenSearch导入大内容DataFrame失败求助

针对你用awswrangler导入含大内容的DataFrame到OpenSearch时无报错但中断的问题，给出以下排查和解决建议：

1. 排查单文档大小是否超限 #

OpenSearch默认限制单文档最大为100MB（通过http.max_content_length配置），先确认大内容文档是否超过这个阈值：

# 计算每个文档内容的字节大小（UTF-8编码）
df['content_bytes'] = df['你的内容列名'].apply(lambda x: len(x.encode('utf-8')))
# 筛选超过100MB的文档
large_docs = df[df['content_bytes'] > 100 * 1024 * 1024]
print(f"发现{len(large_docs)}个超大小文档")

如果存在超大小文档，要么拆分长文档，要么联系集群管理员调整OpenSearch的http.max_content_length参数。

2. 缩小批量导入的尺寸 #

当前设置的bulk_size=1000可能导致单个批量请求总大小过大，触发超时或服务端限制。调小批量参数并重试：

wr.opensearch.index_df(
    client,
    df=df,
    index="indexname",
    id_keys=["index_no"],
    max_retries=5,
    retry_delay=2,  # 增加重试间隔
    bulk_size=100,  # 大幅降低单批次文档数
    chunk_size=50
)

3. 启用详细日志排查隐藏错误 #

默认日志级别无法展示底层请求错误，开启DEBUG日志获取更多细节：

import logging
logging.basicConfig(level=logging.DEBUG)
# 再执行导入代码，查看输出的请求响应细节

4. 单独测试大文档 #

将大文档单独筛选出来，小批量导入以定位具体错误：

# 筛选内容字节数超过50MB的文档
test_df = df[df['content_bytes'] > 50 * 1024 * 1024]
# 小批量导入测试
wr.opensearch.index_df(
    client,
    df=test_df,
    index="indexname",
    id_keys=["index_no"],
    bulk_size=10,
    chunk_size=5
)

5. 改用OpenSearch原生客户端手动处理批量 #

如果awswrangler的封装无法满足需求，直接用opensearch-py客户端实现批量导入，可精准控制超时和捕获错误：

from opensearchpy import OpenSearch, helpers

# 初始化原生客户端
os_client = OpenSearch(
    hosts=[{'host': 'back end url', 'port': 443}],
    http_auth=('username', 'password'),
    use_ssl=True,
    verify_certs=True
)

# 生成批量导入动作
def generate_import_actions(df):
    for _, row in df.iterrows():
        yield {
            "_index": "indexname",
            "_id": row["index_no"],
            "_source": row.to_dict()
        }

# 执行批量导入，设置更长超时和小批量
success_count, failed_items = helpers.bulk(
    os_client,
    generate_import_actions(df),
    chunk_size=100,
    request_timeout=60,  # 延长请求超时到60秒
    raise_on_error=False,
    raise_on_exception=False
)

print(f"成功导入: {success_count} 条")
if failed_items:
    print("失败文档详情:")
    for item in failed_items:
        print(f"ID: {item['index']['_id']}, 错误信息: {item['index']['error']}")

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