使用pandas优化百万级Elasticsearch查询结果导出CSV的问题咨询

性能问题根因 #

当前性能瓶颈完全来自pandas的append操作：pandas的DataFrame/Series为不可变对象，每次调用append方法都会完整复制当前全部存量数据生成新对象，随着积累的数据量上升，单次复制的耗时会呈指数级增长。你日志中ES的请求耗时仅为数百毫秒，两次scroll请求的间隔时间几乎全部消耗在pandas的数据复制操作上。

优化方案 #

1. 低改动高收益方案：列表暂存+一次性转DataFrame #

列表的append是原地修改操作，时间复杂度为O(1)，没有额外的数据复制开销，仅需修改几行代码就能获得十倍以上的性能提升，示例代码如下：

import pandas as pd

# 初始化空列表存储所有行数据
docs_list = []

for hit in scan(elastic_client, index=index, query=query, scroll='20h', clear_scroll=True, size=5000):
    scan_source_data = hit["_source"]
    # 将_id存入数据中，后续可设为索引对齐原有逻辑
    scan_source_data["_id"] = hit["_id"]
    docs_list.append(scan_source_data)

# 全量数据拉取完成后一次性转DataFrame
scan_docs = pd.DataFrame(docs_list).set_index("_id")
scan_docs.to_csv("/tmp/scandocs.csv")

2. 大内存友好方案：分批写入CSV #

如果100万条全量数据无法全部放入内存，可以每攒够固定批次的数据就写入一次CSV，避免全量数据驻留内存：

import pandas as pd

# 每攒够1万条写一次CSV，可根据实际内存情况调整大小
batch_size = 10000
current_batch = []
# 标记是否为首次写入，控制是否写入表头
first_write = True

for hit in scan(elastic_client, index=index, query=query, scroll='20h', clear_scroll=True, size=5000):
    scan_source_data = hit["_source"]
    scan_source_data["_id"] = hit["_id"]
    current_batch.append(scan_source_data)
    
    if len(current_batch) >= batch_size:
        df = pd.DataFrame(current_batch).set_index("_id")
        df.to_csv("/tmp/scandocs.csv", mode='a', header=first_write, index=True)
        # 清空批次缓存，更新表头标记
        current_batch = []
        first_write = False

# 写入最后一批不足批次大小的剩余数据
if current_batch:
    df = pd.DataFrame(current_batch).set_index("_id")
    df.to_csv("/tmp/scandocs.csv", mode='a', header=first_write, index=True)

3. 可选调优项 #

• 可以将scan的size参数调整到10000，减少请求ES的次数，只要单条数据体积不大，这个大小不会产生额外性能问题
• 如果字段格式固定，也可以直接用Python内置的csv模块写入文件，跳过pandas转换步骤，性能可以进一步提升

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