Elasticsearch全产品聚合查询及Pandas批量生成每日需求DataFrame方法

1. 可以一次性对所有产品执行查询聚合操作 #

直接移除查询条件中针对product_id的过滤，同时调整聚合结构：先按product_id分桶，再在每个产品桶内按day做日期聚合，一次请求就能获取所有产品的每日统计数据，效率远高于循环单个产品查询。

2. 无需循环处理单个产品，修改聚合请求即可生成目标DataFrame #

直接调整Elasticsearch的聚合逻辑，一次性拉取所有产品的每日数据，再解析成Pandas DataFrame，代码示例如下：

修改后的ES查询体 #

search_body_statistics = {
    "size": 0,
    "aggs": {
        "products": {
            "terms": {
                "field": "product_id",
                "size": 10000  # 取值需覆盖所有产品ID数量
            },
            "aggs": {
                "countPerDay": {
                    "terms": {
                        "field": "day",
                        "size": 10000,
                        "order": {
                            "_key": "asc"
                        }
                    }
                }
            }
        }
    }
}

解析结果生成DataFrame #

result_stat = es.search(index="sales", body=search_body_statistics)

# 逐层解析聚合结果
data_records = []
for product_bucket in result_stat['aggregations']['products']['buckets']:
    current_product = product_bucket['key']
    for day_bucket in product_bucket['countPerDay']['buckets']:
        data_records.append({
            'product_id': current_product,
            'day': day_bucket['key'],
            'demand': day_bucket['doc_count']
        })

# 生成包含所有产品每日需求的DataFrame
df_all_products = pd.DataFrame(data_records)
print(df_all_products)

补充说明 #

• 如果产品数量极大，terms聚合的size参数需设置为足够容纳所有产品ID，或者改用composite聚合实现分页查询，避免内存溢出。
• 此方法仅需一次ES请求，相比循环单个产品查询，性能提升显著，尤其在产品数量较多的场景下。

内容的提问来源于stack exchange，提问作者RB Vries de