有一种方法可以使用scripted_metric
聚合和sum_bucket
管道聚合的组合来解决。脚本化的指标聚合有点复杂,但主要思想是允许您提供自己的存储算法并从中吐出单个指标。
在您的情况下,您要做的是找出每个商店的最新数量,然后对这些商店数量求和。解决方案如下所示,我将在下面解释一些细节:
POST inventory-local/_search
{
"size": 0,
"aggs": {
"bystore": {
"terms": {
"field": "store.keyword",
"size": 10000
},
"aggs": {
"latest_quantity": {
"scripted_metric": {
"init_script": "params._agg.quantities = new TreeMap()",
"map_script": "params._agg.quantities.put(doc.datetime.date, [doc.datetime.date.millis, doc.quantity.value])",
"combine_script": "return params._agg.quantities.lastEntry().getValue()",
"reduce_script": "def maxkey = 0; def qty = 0; for (a in params._aggs) {def currentKey = a[0]; if (currentKey > maxkey) {maxkey = currentKey; qty = a[1]} } return qty;"
}
}
}
},
"sum_latest_quantities": {
"sum_bucket": {
"buckets_path": "bystore>latest_quantity.value"
}
}
}
}
请注意,为了使其正常工作,您需要script.painless.regex.enabled: true
在elasticsearch.yml
配置文件中进行设置。
在init_script
创建TreeMap
每个碎片。使用日期/数量的映射map_script
填充TreeMap
每个分片上的。我们在地图中输入的值在单个字符串中包含时间戳和数量。我们稍后将需要该时间戳记reduce_script
。在combine_script
简单地采取的最后一个值TreeMap
,因为这是给定的碎片最新的量。大部分工作位于reduce_script
。我们迭代每个分片的所有最新数量,并返回最新的数量。
此时,我们为每个商店提供了最新数量。剩下要做的就是使用sum_bucket
管道聚合来求和每个存储量。在那里,您得到17的结果。
响应如下所示:
"aggregations": {
"bystore": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "01",
"doc_count": 2,
"latest_quantity": {
"value": 6
}
},
{
"key": "02",
"doc_count": 2,
"latest_quantity": {
"value": 11
}
}
]
},
"sum_latest_quantities": {
"value": 17
}
}