如何使用Python将OpenSearch数据导入Excel 解决10k+记录与增量加载问题

超10k条记录批量插入实现 #

OpenSearch 单条插入性能极低，且单次HTTP请求存在条数、包体大小限制，直接提交超10k条单请求会直接被拦截，使用官方bulk批量API做分片提交即可解决：

• 单批次提交条数建议控制在1000~5000条区间，单批次总payload大小不超过10MB，可根据集群节点配置压测后调整
• 原有客户端配置中已开启http_compress = True，批量场景下该配置可降低60%以上的传输耗时，无需修改
• 批量提交需增加失败重试、错误条目收集逻辑，避免部分写入失败导致数据不一致

批量插入参考代码：

from opensearchpy import OpenSearch, helpers
import pandas as pd

# 沿用原有客户端初始化逻辑
host = ''
port = 
auth = ('', '') 
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True,
    http_auth = auth,
    use_ssl = True,
    verify_certs = False,
    ssl_assert_hostname = False,
    ssl_show_warn = False
)
index_name = "你的索引名"

# 待插入的超10k条数据生成器，可替换为任意数据源的迭代逻辑
def generate_insert_docs(source_data):
    for row in source_data:
        yield {
            "_index": index_name,
            "_id": row.get("唯一主键字段"), # 建议指定业务主键，避免重复插入
            "_source": row
        }

# 批量自动分片提交
success_count, failed_items = helpers.bulk(
    client,
    generate_insert_docs(your_source_data),
    chunk_size=2000, # 单批次提交条数
    max_retries=3, # 失败重试次数
    request_timeout=120
)
print(f"插入成功{success_count}条，失败{len(failed_items)}条")

增量加载功能配置搭建 #

增量加载核心是记录同步位点，每次只拉取位点之后的新增/变更数据，避免重复全量拉取，两种常用实现方案：

方案1：时间戳游标增量（推荐，成本最低） #

要求索引中所有写入数据都携带create_time/update_time类的时间戳字段，每次同步后记录本次同步的最大时间戳作为下次同步的起点：

1. 首次同步时全量拉取所有数据，记录当前同步到的最大时间戳last_sync_time持久化到本地/数据库
2. 后续每次同步只查询update_time > last_sync_time的数据，同步完成后更新last_sync_time
3. 数据量较大时配合search_after做深度分页，避免from+size的1w条上限问题

参考代码片段：

# 读取上次同步位点，首次同步可设为0
last_sync_time = load_last_sync_time_from_storage()

query = {
    "query": {
        "range": {
            "update_time": {
                "gt": last_sync_time
            }
        }
    },
    "sort": [{"update_time": "asc"}, {"_id": "asc"}], # sort字段必须包含唯一值，避免分页丢数据
    "size": 2000
}

all_docs = []
# 首次查询
resp = client.search(index=index_name, body=query)
all_docs.extend(resp["hits"]["hits"])
# 循环翻页拉取所有增量数据
while len(resp["hits"]["hits"]) > 0:
    last_sort = resp["hits"]["hits"][-1]["sort"]
    query["search_after"] = last_sort
    resp = client.search(index=index_name, body=query)
    all_docs.extend(resp["hits"]["hits"])

# 数据处理完成后更新最新同步位点
if all_docs:
    new_sync_time = max([doc["_source"]["update_time"] for doc in all_docs])
    save_last_sync_time_to_storage(new_sync_time)

方案2：Scroll游标全量增量（适合无时间戳字段场景） #

原有示例代码中只调用了一次search接口，没有循环读取scroll_id对应的后续批次，所以最多只能拿到1w条数据，补全scroll循环逻辑即可拉取全量数据，适合首次全量同步场景：

注意：scroll游标会占用集群内存，同步完成后要主动调用clear_scroll接口释放资源，不要长期持有scroll_id

query = {"query": {"match_all": {}}}
resp = client.search(
    body=query,
    index=index_name,
    scroll="2m", # 游标有效期，根据单批次处理时长调整
    size=2000
)
scroll_id = resp["_scroll_id"]
all_docs = resp["hits"]["hits"]

while len(resp["hits"]["hits"]) > 0:
    resp = client.scroll(scroll_id=scroll_id, scroll="2m")
    scroll_id = resp["_scroll_id"]
    all_docs.extend(resp["hits"]["hits"])

# 同步完成后清理游标
client.clear_scroll(scroll_id=scroll_id)
# 后续处理逻辑，比如转存Excel、写入下游等
df = pd.json_normalize(all_docs)
df.to_excel("export_dataframe.xlsx", index=False)

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