Python脚本Elasticsearch批量写入怎么做

本文目录导读：

1. 使用 elasticsearch 官方库的 bulk 方法
2. 使用 bulk 方法（手动构造）
3. 使用 parallel_bulk 并行写入
4. 使用 streaming_bulk 流式写入
5. 优化配置示例
6. 错误处理和重试机制
7. 使用 DataFrame 批量写入
8. 关键参数说明
9. 注意事项

我来为您介绍Python中Elasticsearch批量写入的几种方法：

使用 elasticsearch 官方库的 bulk 方法

安装依赖

pip install elasticsearch

基础示例

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
# 连接Elasticsearch
es = Elasticsearch(['http://localhost:9200'])
def generate_data():
    """生成要写入的数据"""
    for i in range(1000):
        yield {
            "_index": "my_index",
            "_id": i,  # 可选，如果不指定会自动生成
            "_source": {
                "title": f"Document {i}",
                "content": f"Content for document {i}",
                "timestamp": "2024-01-01"
            }
        }
# 批量写入
success, failed = bulk(es, generate_data())
print(f"成功: {success}, 失败: {failed}")

使用 bulk 方法（手动构造）

from elasticsearch import Elasticsearch
import json
es = Elasticsearch(['http://localhost:9200'])
# 准备批量数据
actions = []
for i in range(1000):
    action = {
        "_index": "my_index",
        "_id": i,
        "_source": {
            "field1": f"value_{i}",
            "field2": i * 10
        }
    }
    actions.append(action)
# 批量写入
from elasticsearch.helpers import bulk
success, failed = bulk(es, actions)

使用 parallel_bulk 并行写入

from elasticsearch import Elasticsearch
from elasticsearch.helpers import parallel_bulk
es = Elasticsearch(['http://localhost:9200'])
def generate_bulk_data():
    for i in range(10000):
        yield {
            "_index": "my_index",
            "_source": {
                "title": f"Document {i}",
                "value": i
            }
        }
# 并行批量写入
for success, info in parallel_bulk(es, generate_bulk_data(), thread_count=4):
    if not success:
        print(f"写入失败: {info}")

使用 streaming_bulk 流式写入

from elasticsearch import Elasticsearch
from elasticsearch.helpers import streaming_bulk
es = Elasticsearch(['http://localhost:9200'])
def data_generator():
    for i in range(5000):
        yield {
            "_index": "my_index",
            "_source": {
                "title": f"Document {i}",
                "content": f"Content {i}",
                "timestamp": "2024-01-01"
            }
        }
# 流式批量写入
ok_count = 0
for ok, result in streaming_bulk(es, data_generator()):
    if not ok:
        print(f"写入失败: {result}")
    else:
        ok_count += 1
print(f"成功写入: {ok_count} 条")

优化配置示例

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
# 配置连接池和超时
es = Elasticsearch(
    ['http://localhost:9200'],
    max_retries=3,
    retry_on_timeout=True,
    timeout=30,
    max_connections=10
)
# 批量写入配置
actions = []
for i in range(10000):
    actions.append({
        "_index": "my_index",
        "_source": {
            "field": f"value_{i}",
            "timestamp": "2024-01-01"
        }
    })
# 使用bulk参数优化
success, failed = bulk(
    es,
    actions,
    chunk_size=500,  # 每批500条
    max_retries=3,   # 最大重试次数
    raise_on_error=False,  # 遇到错误不停止
    request_timeout=60  # 请求超时时间
)
print(f"成功: {success}, 失败: {failed}")

错误处理和重试机制

from elasticsearch import Elasticsearch, ElasticsearchException
from elasticsearch.helpers import bulk
import time
es = Elasticsearch(['http://localhost:9200'])
def bulk_with_retry(es, actions, max_retries=3):
    retries = 0
    while retries < max_retries:
        try:
            success, failed = bulk(
                es, 
                actions,
                raise_on_error=False
            )
            if failed > 0:
                print(f"有 {failed} 条写入失败，重试中...")
                actions = [a for a in actions if a.get('_id')]  # 重新构造失败的数据
                retries += 1
                time.sleep(1)
            else:
                return success, failed
        except ElasticsearchException as e:
            print(f"Elasticsearch错误: {e}")
            retries += 1
            time.sleep(2 ** retries)  # 指数退避
    return 0, len(actions)
# 使用
actions = []
for i in range(1000):
    actions.append({
        "_index": "my_index",
        "_source": {"field": f"value_{i}"}
    })
success, failed = bulk_with_retry(es, actions)

使用 DataFrame 批量写入

import pandas as pd
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import numpy as np
# 创建DataFrame
df = pd.DataFrame({: [f'Doc {i}' for i in range(1000)],
    'content': np.random.randn(1000),
    'category': ['A', 'B', 'C'] * 333 + ['A']
})
# 转换为Elasticsearch格式
def df_to_bulk(df, index_name):
    for i, row in df.iterrows():
        yield {
            "_index": index_name,
            "_id": i,
            "_source": row.to_dict()
        }
es = Elasticsearch(['http://localhost:9200'])
success, failed = bulk(es, df_to_bulk(df, "my_index"))

关键参数说明

• chunk_size: 每批处理的数据量（默认500）
• max_retries: 失败重试次数
• raise_on_error: 错误时是否抛出异常
• request_timeout: 请求超时时间
• thread_count: 并行写入线程数（parallel_bulk）

注意事项

1. 批量大小: 建议每批500-1000条
2. 索引设置: 批量写入前关闭刷新和副本可以提高性能
3. 错误处理: 务必添加异常处理机制
4. 监控: 监控写入速度和失败率
5. 索引模板: 预定义索引映射可以提高性能

选择合适的批量写入方法取决于您的具体需求和数据量大小。