18

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

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

  1. 使用聚合：示例：https : //found.no/play/gist/8124563 如果您必须按一个字段分组，并且需要一些额外的字段，看起来很有用。

  2. 在一个 Facet 中使用多个字段（不起作用）：示例：https ://found.no/play/gist/1aa44e2114975384a7c2 在这里我们失去了不同字段之间的关系。

  3. 构建有趣的方面： ​​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 个。

4

4 回答 4

33

聚合 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，则不需要它）

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
于 2014-01-23T07:35:55.220 回答
10

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

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

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

于 2019-09-11T07:34:38.013 回答
5

我认为一些开发人员肯定会在 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;
于 2017-03-07T11:51:10.773 回答
1

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

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

{
"aggregations": {
    "VALUE1AGG": {
      "terms": {
        "field": "VALUE1",
      },
      "aggregations": {
        "VALUE2AGG": {
           "terms": {
             "field": "VALUE2",
          }
        }
      }
    }
  }
}
于 2015-03-17T11:03:49.263 回答