将Elasticsearch Python聚合结果转为可读JSON/CSV/DataFrame的方法咨询

一、格式化输出可读JSON #

Elasticsearch返回的res是Python字典结构，直接用json模块格式化后就能得到层级清晰的JSON：

import json

# 格式化输出，缩进4空格，保留中文
print(json.dumps(res, indent=4, ensure_ascii=False))

二、扁平化嵌套聚合结果 #

你的聚合是多层嵌套结构（时间桶→目的国→航空公司→目的国→延误状态），直接查看仍会很繁琐，需要把嵌套层级展开为一维数据列表：

def flatten_aggregations(agg_result):
    flat_data = []
    # 遍历时间桶（对应聚合"2"的结果）
    for time_bucket in agg_result['aggregations']['2']['buckets']:
        time_key = time_bucket['key_as_string']
        time_total = time_bucket['doc_count']
        # 遍历第一层目的国（对应聚合"3"的结果）
        for dest_bucket in time_bucket['3']['buckets']:
            dest_country = dest_bucket['key']
            dest_total = dest_bucket['doc_count']
            # 遍历航空公司（对应聚合"4"的结果）
            for carrier_bucket in dest_bucket['4']['buckets']:
                carrier = carrier_bucket['key']
                carrier_total = carrier_bucket['doc_count']
                # 遍历第二层目的国（对应聚合"5"的结果，此处与上层目的国字段重复，疑似笔误）
                for dest2_bucket in carrier_bucket['5']['buckets']:
                    dest_country2 = dest2_bucket['key']
                    dest2_total = dest2_bucket['doc_count']
                    # 遍历航班延误状态（对应聚合"6"的结果）
                    for delay_bucket in dest2_bucket['6']['buckets']:
                        flight_delay = delay_bucket['key']
                        delay_total = delay_bucket['doc_count']
                        # 整合所有维度与统计值
                        flat_data.append({
                            '时间': time_key,
                            '时间桶总记录数': time_total,
                            '目的国': dest_country,
                            '目的国总记录数': dest_total,
                            '航空公司': carrier,
                            '航空公司总记录数': carrier_total,
                            '二次目的国': dest_country2,
                            '二次目的国总记录数': dest2_total,
                            '航班延误状态': flight_delay,
                            '延误状态记录数': delay_total
                        })
    return flat_data

# 获取扁平化后的结构化数据
flat_result = flatten_aggregations(res)

三、转为Pandas DataFrame #

基于扁平化数据，直接生成DataFrame即可实现表格化展示：

import pandas as pd

df = pd.DataFrame(flat_result)
# 查看前5行数据
print(df.head())

四、导出为CSV文件 #

利用DataFrame的内置方法直接导出为CSV：

# 导出CSV，去除索引，保留中文编码
df.to_csv('es_flight_agg_result.csv', index=False, encoding='utf-8-sig')

内容的提问来源于stack exchange，提问作者jack sumatra