elasticsearch group-by 多个字段答案

【问题标题】：elasticsearch group-by multiple fieldselasticsearch group-by 多个字段
【发布时间】：2013-12-25 17:03:53
【问题描述】：

我正在寻找在 elasticsearch 中对数据进行分组的最佳方法。 Elasticsearch 不支持 sql 中的“分组依据”之类的内容。

假设我有 1k 个类别和数百万种产品。您认为渲染完整类别树的最佳方式是什么？当然，您需要一些元数据（图标、链接目标、seo 标题等）和类别的自定义排序。

使用聚合：示例：https://found.no/play/gist/8124563 如果您必须按一个字段分组，并且需要一些额外的字段，看起来很有用。
在 Facet 中使用多个字段（不起作用）：示例：https://found.no/play/gist/1aa44e2114975384a7c2 这里我们失去了不同字段之间的关系。
构建有趣的 Facet： https://found.no/play/gist/8124810

例如，使用这 3 个“解决方案”构建类别树很糟糕。解决方案 1 可能有效（ES 1 目前不稳定）解决方案 2 不起作用解决方案 3 很痛苦，因为感觉很难看，你需要准备很多数据，而且方面都炸了。

也许另一种选择是不在 ES 中存储任何类别数据，只存储 id https://found.no/play/gist/a53e46c91e2bf077f2e1

然后你可以从另一个系统中获取相关的类别，比如 redis、memcache 或数据库。

这会以干净的代码结束，但性能可能会成为问题。例如，从 Memcache / Redis / 数据库加载 1k 类别可能会很慢。另一个问题是同步 2 个数据库比同步一个更难。

你如何处理这些问题？

对不起，我不能在一篇文章中发布超过 2 个链接。

【问题讨论】：

标签： elasticsearch group-by elasticsearch-query

【解决方案1】：

聚合 API 允许使用子聚合按多个字段进行分组。假设您要按字段field1、field2 和field3 进行分组：

{
  "aggs": {
    "agg1": {
      "terms": {
        "field": "field1"
      },
      "aggs": {
        "agg2": {
          "terms": {
            "field": "field2"
          },
          "aggs": {
            "agg3": {
              "terms": {
                "field": "field3"
              }
            }
          }          
        }
      }
    }
  }
}

当然，这可以在任意多个领域进行。

更新：
为了完整起见，以下是上述查询的输出外观。下面还有用于生成聚合查询并将结果展平为字典列表的 python 代码。

{
  "aggregations": {
    "agg1": {
      "buckets": [{
        "doc_count": <count>,
        "key": <value of field1>,
        "agg2": {
          "buckets": [{
            "doc_count": <count>,
            "key": <value of field2>,
            "agg3": {
              "buckets": [{
                "doc_count": <count>,
                "key": <value of field3>
              },
              {
                "doc_count": <count>,
                "key": <value of field3>
              }, ...
              ]
            },
            {
            "doc_count": <count>,
            "key": <value of field2>,
            "agg3": {
              "buckets": [{
                "doc_count": <count>,
                "key": <value of field3>
              },
              {
                "doc_count": <count>,
                "key": <value of field3>
              }, ...
              ]
            }, ...
          ]
        },
        {
        "doc_count": <count>,
        "key": <value of field1>,
        "agg2": {
          "buckets": [{
            "doc_count": <count>,
            "key": <value of field2>,
            "agg3": {
              "buckets": [{
                "doc_count": <count>,
                "key": <value of field3>
              },
              {
                "doc_count": <count>,
                "key": <value of field3>
              }, ...
              ]
            },
            {
            "doc_count": <count>,
            "key": <value of field2>,
            "agg3": {
              "buckets": [{
                "doc_count": <count>,
                "key": <value of field3>
              },
              {
                "doc_count": <count>,
                "key": <value of field3>
              }, ...
              ]
            }, ...
          ]
        }, ...
      ]
    }
  }
}

以下 python 代码在给定字段列表的情况下执行分组。如果您指定 include_missing=True，它还包括缺少某些字段的值组合（如果您有 2.0 版的 Elasticsearch，则不需要它，感谢 this）

def group_by(es, fields, include_missing):
    current_level_terms = {'terms': {'field': fields[0]}}
    agg_spec = {fields[0]: current_level_terms}

    if include_missing:
        current_level_missing = {'missing': {'field': fields[0]}}
        agg_spec[fields[0] + '_missing'] = current_level_missing

    for field in fields[1:]:
        next_level_terms = {'terms': {'field': field}}
        current_level_terms['aggs'] = {
            field: next_level_terms,
        }

        if include_missing:
            next_level_missing = {'missing': {'field': field}}
            current_level_terms['aggs'][field + '_missing'] = next_level_missing
            current_level_missing['aggs'] = {
                field: next_level_terms,
                field + '_missing': next_level_missing,
            }
            current_level_missing = next_level_missing

        current_level_terms = next_level_terms

    agg_result = es.search(body={'aggs': agg_spec})['aggregations']
    return get_docs_from_agg_result(agg_result, fields, include_missing)


def get_docs_from_agg_result(agg_result, fields, include_missing):
    current_field = fields[0]
    buckets = agg_result[current_field]['buckets']
    if include_missing:
        buckets.append(agg_result[(current_field + '_missing')])

    if len(fields) == 1:
        return [
            {
                current_field: bucket.get('key'),
                'doc_count': bucket['doc_count'],
            }
            for bucket in buckets if bucket['doc_count'] > 0
        ]

    result = []
    for bucket in buckets:
        records = get_docs_from_agg_result(bucket, fields[1:], include_missing)
        value = bucket.get('key')
        for record in records:
            record[current_field] = value
        result.extend(records)

    return result

【讨论】：

【解决方案2】：

您可以按如下方式使用复合聚合查询。如果桶的数量超过 ES 的正常值，这种类型的查询也会对结果进行分页。通过使用“之后”字段，您可以访问其余的存储桶：

"aggs": {
    "my_buckets": {
      "composite": {
        "sources": [
          {
            "field1": {
              "terms": {
                "field": "field1"
              }
            }
          },
          {
            "field2": {
              "terms": {
                "field": "field2"
              }
            }
          },
         {
            "field3": {
              "terms": {
                "field": "field3"
              }
            }
          },
        ]
      }
    }
  }

您可以在ES页面bucket-composite-aggregation找到更多详细信息。

【讨论】：

es 6.1 以后支持复合聚合

【解决方案3】：

我认为一些开发人员肯定会在 Spring DATA ES 和 JAVA ES API 中寻找相同的实现。

请找到：-

List<FieldObject> fieldObjectList = Lists.newArrayList();
    SearchQuery aSearchQuery = new NativeSearchQueryBuilder().withQuery(matchAllQuery()).withIndices(indexName).withTypes(type)
            .addAggregation(
                    terms("ByField1").field("field1").subAggregation(AggregationBuilders.terms("ByField2").field("field2")
                            .subAggregation(AggregationBuilders.terms("ByField3").field("field3")))
                    )
            .build();
    Aggregations aField1Aggregations = elasticsearchTemplate.query(aSearchQuery, new ResultsExtractor<Aggregations>() {
        @Override
        public Aggregations extract(SearchResponse aResponse) {
            return aResponse.getAggregations();
        }
    });
    Terms aField1Terms = aField1Aggregations.get("ByField1");
    aField1Terms.getBuckets().stream().forEach(aField1Bucket -> {
        String field1Value = aField1Bucket.getKey();
        Terms aField2Terms = aField1Bucket.getAggregations().get("ByField2");

        aField2Terms.getBuckets().stream().forEach(aField2Bucket -> {
            String field2Value = aField2Bucket.getKey();
            Terms aField3Terms = aField2Bucket.getAggregations().get("ByField3");

            aField3Terms.getBuckets().stream().forEach(aField3Bucket -> {
                String field3Value = aField3Bucket.getKey();
                Long count = aField3Bucket.getDocCount();

                FieldObject fieldObject = new FieldObject();
                fieldObject.setField1(field1Value);
                fieldObject.setField2(field2Value);
                fieldObject.setField3(field3Value);
                fieldObject.setCount(count);
                fieldObjectList.add(fieldObject);
            });
        });
    });

同样需要导入：-

import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.elasticsearch.search.aggregations.AggregationBuilders.terms; 
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.collect.Lists;
import org.elasticsearch.index.query.FilterBuilder;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.TermFilterBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.ResultsExtractor;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;

【讨论】：

【解决方案4】：

子聚合是您所需要的 .. 虽然这在文档中从未明确说明过，但 structuring aggregations 可以隐含地找到它

这将产生子聚合，就好像查询被更高聚合的结果过滤一样。它实际上看起来好像这就是那里发生的事情。

{
"aggregations": {
    "VALUE1AGG": {
      "terms": {
        "field": "VALUE1",
      },
      "aggregations": {
        "VALUE2AGG": {
           "terms": {
             "field": "VALUE2",
          }
        }
      }
    }
  }
}

【讨论】：