如何聚合直到在 ElasticSearch 中达到某个值？答案

【问题标题】：How to aggregate until a certain value is reached in ElasticSearch?如何聚合直到在 ElasticSearch 中达到某个值？
【发布时间】：2019-10-03 12:35:37
【问题描述】：

我想按“金额”字段聚合文档列表（每个文档都有两个字段 - 时间戳和金额），直到达到某个值。例如，我想获取按时间戳排序的文档列表，总数量等于 100。是否可以在一个查询中完成？

这是我的查询，它返回总金额 - 我想在这里添加一个条件，以在达到某个值时停止聚合。

{
"query": {
    "bool": {
        "filter": [
            {
                "range": {
                    "timestamp": {
                        "gte": 1525168583
                    }
                }
            }
        ]
    }
},
"aggs": {
    "total_amount": {
        "sum": {
            "field": "amount"
        }
    }
},
"sort": [
    "timestamp"
],
"size": 10000
}

谢谢

【问题讨论】：

请注意sort 和size 与聚合无关，仅与返回的命中有关。所以不清楚你想看到什么结果。你能展示一些你想看的模拟例子吗？
是的，我知道，抱歉，您可以忽略它们。
不过，我还是希望您用一个示例来解释您希望看到的结果。您希望每小时/每天/...的总数不超过 100 吗？或者获取按时间戳排序的文档集，总数量达到100？
我想得到一个按时间戳排序的文档列表，总数量达到 100 个。
答案是：不，不可能。聚合不会影响查询本身、排序或限制结果集的大小。它们应用于与查询匹配的所有文档（与“大小”值无关）。

标签： elasticsearch elasticsearch-aggregation elasticsearch-dsl-py

【解决方案1】：

完全有可能将function_score scripting 用于模拟排序、filter aggs 用于范围 gte 查询和健康数量的scripted_metric aggs 组合使用以将总和限制在一定数量：

PUT summation
{
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "epoch_second"
      }
    }
  }
}
POST summation/_doc
{
  "context": "newest",
  "timestamp": 1587049128,
  "amount": 20
}

POST summation/_doc
{
  "context": "2nd newest",
  "timestamp": 1586049128,
  "amount": 30
}

POST summation/_doc
{
  "context": "3rd newest",
  "timestamp": 1585049128,
  "amount": 40
}

POST summation/_doc
{
  "context": "4th newest",
  "timestamp": 1585049128,
  "amount": 30
}

GET summation/_search
{
  "size": 0,
  "aggs": {
    "filtered_agg": {
      "filter": {
        "bool": {
          "must": [
            {
              "range": {
                "timestamp": {
                  "gte": 1585049128
                }
              }
            },
            {
              "function_score": {
                "query": {
                  "match_all": {}
                },
                "script_score": {
                  "script": {
                    "source": "return (params['now'] - doc['timestamp'].date.toMillis())",
                    "params": {
                      "now": 1587049676
                    }
                  }
                }
              }
            }
          ]
        }
      },
      "aggs": {
        "limited_sum": {
          "scripted_metric": {
            "init_script": """
                state['my_hash'] = new HashMap();
                state['my_hash'].put('sum', 0);
                state['my_hash'].put('docs', new ArrayList());
            """,
            "map_script": """
              if (state['my_hash']['sum'] <= 100) {
                state['my_hash']['sum'] += doc['amount'].value;
                state['my_hash']['docs'].add(doc['context.keyword'].value);
              }
            """,
            "combine_script": "return state['my_hash']",
            "reduce_script": "return states[0]"
          }
        }
      }
    }
  }
}

屈服

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 4,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "filtered_agg" : {
      "meta" : { },
      "doc_count" : 4,
      "limited_sum" : {
        "value" : {
          "docs" : [
            "newest",
            "2nd newest",
            "3rd newest",
            "4th newest"
          ],
          "sum" : 120
        }
      }
    }
  }
}

我在这里选择只返回 doc.contexts，但您可以对其进行调整以检索您喜欢的任何内容 - 无论是 ID、金额等。

【讨论】：