AsyncElasticsearch 异步批量操作实践指南

import asyncio
from elasticsearch import AsyncElasticsearch, helpers

# 假设您的Elasticsearch运行在本地，并使用默认端口
# 实际应用中，请替换为您的ES集群地址
ES_HOST = "http://localhost:9200"
INDEX_NAME = "my_async_index"

async def perform_async_bulk_indexing():
    # 初始化 AsyncElasticsearch 客户端
    # 建议使用 async with 语句管理客户端生命周期
    async with AsyncElasticsearch(ES_HOST) as es:
        # 1. 检查并创建索引（如果不存在）
        if not await es.indices.exists(index=INDEX_NAME):
            await es.indices.create(index=INDEX_NAME)
            print(f"索引 '{INDEX_NAME}' 已创建。")
        else:
            print(f"索引 '{INDEX_NAME}' 已存在。")

        # 2. 准备要批量操作的数据
        # 每个字典代表一个操作，通常包含 "_index", "_id", "_source"
        documents = [
            {
                "_index": INDEX_NAME,
                "_id": "doc1",
                "_source": {"title": "Async Bulk Operations", "author": "Alice", "views": 100}
            },
            {
                "_index": INDEX_NAME,
                "_id": "doc2",
                "_source": {"title": "Elasticsearch in Python", "author": "Bob", "views": 150}
            },
            {
                "_index": INDEX_NAME,
                "_id": "doc3",
                "_source": {"title": "FastAPI with Elasticsearch", "author": "Charlie", "views": 200}
            },
            {
                "_index": INDEX_NAME,
                "_id": "doc4",
                "_source": {"title": "Optimizing Async Applications", "author": "Alice", "views": 120}
            },
        ]

        print(f"\n开始批量索引 {len(documents)} 篇文档...")

        # 3. 调用 helpers.async_bulk 执行批量操作
        # actions 参数可以是一个生成器或列表
        # yield_ok=False 表示只返回失败的文档信息，默认是True
        success_count, failed_actions = await helpers.async_bulk(
            es,
            documents,
            index=INDEX_NAME, # 可以在这里指定默认索引，也可以在每个文档中指定
            chunk_size=500,    # 每次发送到ES的文档数量
            max_retries=3,     # 失败后重试次数
            initial_backoff=2, # 初始重试等待时间（秒）
            max_backoff=60,    # 最大重试等待时间（秒）
            raise_on_error=False, # 遇到错误时不抛出异常，而是返回失败列表
            raise_on_exception=False # 遇到异常时不抛出异常，而是返回失败列表
        )

        print(f"\n批量操作完成。")
        print(f"成功索引文档数量: {success_count}")

        # 4. 处理失败的文档
        if failed_actions:
            print(f"以下文档未能成功索引 ({len(failed_actions)} 篇):")
            for item in failed_actions:
                print(f"  - {item}")
        else:
            print("所有文档均成功索引。")

        # 5. 刷新索引并查询验证
        await es.indices.refresh(index=INDEX_NAME)
        search_result = await es.search(index=INDEX_NAME, query={"match_all": {}})
        print(f"\n索引 '{INDEX_NAME}' 中当前文档总数: {search_result['hits']['total']['value']}")

if __name__ == "__main__":
    asyncio.run(perform_async_bulk_indexing())