摘要：實現(xiàn)聚合通過實現(xiàn)聚合，詳情可見文檔有時候查詢數(shù)據(jù)的時候可能需要實現(xiàn)多字段的功能，例如要實現(xiàn)多個維度的聚合，需要嵌套的查詢語句設(shè)置一個大的分桶數(shù)，防止一次統(tǒng)計不完整用函數(shù)構(gòu)建聚合語句的部分處理得到的數(shù)據(jù)，將其組織成

es通過agg實現(xiàn)聚合，詳情可見 es文檔

有時候查詢es數(shù)據(jù)的時候可能需要實現(xiàn)多字段group by的功能，例如：

SELECT sum(item_count) from A group by field1, field2, field3

要實現(xiàn)多個維度的聚合，需要嵌套的agg查詢語句:

{
    "query": {
    },
    "aggs": {
        "field1": {
            "terms": {
                "field": "field1",
                "size": 2147483647 #設(shè)置一個大的分桶數(shù)，防止一次統(tǒng)計不完整
            },
            "aggs": {
                "field2": {
                    "terms": {
                        "field": "field2",
                        "size": 2147483647
                    },
                    "aggs": {
                        "field3": {
                            "terms": {
                                "field": "field3",
                                "size": 2147483647
                            },
                            "aggs": {
                                "sum_field": {
                                    "sum": {
                                        "field": "sum_field"
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    },
    "size": 0
}

用函數(shù)構(gòu)建聚合語句的agg部分：

def build_query_aggs(fields, sum_field):
    agg_data = {}
    curr_field = agg_data
    for item in fields:
        curr_field[item] = {
            "terms": {
                "field": item,
                "size": 2147483647
            },
            "aggs": {}
        }
        curr_field = curr_field[item]["aggs"]
    curr_field[sum_field] = {
        "sum": {
                "field": sum_field
            }
    }
    return agg_data

處理得到的數(shù)據(jù)，將其組織成list：

def build_es_aggs_data(data, fields, sum_field):
    curr_field = None
    res_data = []
    if len(fields) > 0:
        curr_field = fields[0]
    else:
        return
    curr_buckets = data[curr_field]["buckets"]
    for item in curr_buckets:
        if len(fields) == 1:
            curr_data= {}
            curr_data[curr_field] = item["key"]
            curr_data[sum_field] = item[sum_field]["value"]
            res_data.append(curr_data)
        else:
            pre_data = deepcopy(build_es_aggs_data(item, fields[1:], sum_field))
            for pre_item in pre_data:
                pre_item[curr_field] = item["key"]
                res_data.append(pre_item)
    return res_data