elasticsearch - 对TOP聚合进行过滤-Elasticsearch 5.6

Question

注意:这种查询在2到3年前曾被问过，但没有令人满意的答案。我在这里发布我的具体问题。希望有人提出一些好的解决方案。

我面临着从elasticsearch获取所需记录的挑战。我们严格需要对TOP聚合返回的结果进行过滤。无论如何，以下是我的情况:

给定:我们有一个名为“service”的实体，其属性如下:

{
    "id": "servicer-id-1",
    "status": "OPEN",         // These may be CLOSED, RESOLVED
    "timeRaised": "2019-03-21T15:09:17.015Z",
    "timeChanged": "2019-03-21T15:09:17.015Z"
}

我有一个 flex 索引，上述服务中的任何更改都存储为整个服务文档(一种服务历史)。有多个具有相同ID的服务。我们每次都更新timeChanges属性。

索引中有数百万个服务文档。

问题陈述:我们需要特定的服务，这些服务是在给定时间范围(timeChanged)和OPEN状态下的最新状态。

我做了什么:
我在下面的查询中使用了10000 bacth大小的滚动API来解决我们的问题:

{
  "size" : 1000,   //given by user
  "query" : {
    "constant_score" : {
      "filter" : {
        "bool" : {
          "must" : [
            {
              "range" : {
                "timeChanged" : {
                  "from" : 1552940830000,
                  "to" : 1553498830000,
                  "include_lower" : true,
                  "include_upper" : true,
                  "boost" : 1.0
                }
              }
            }
          ],
          "disable_coord" : false,
          "adjust_pure_negative" : true,
          "boost" : 1.0
        }
      },
      "boost" : 1.0
    }
  },
  "post_filter": {
    "bool": {
        "must": [{
            {
                "constant_score": {
                    "filter": {
                        "terms": {
                            "status": ["OPEN"],
                            "boost": 1.0
                        }
                    },
                    "boost": 1.0
                }
            }
        }],
      "disable_coord" : false,
      "adjust_pure_negative" : true,
      "boost" : 1.0
    }
  },
  "_source" : false,
  "aggregations" : {
    "by_serviceId" : {
      "terms" : {
        "field" : "id",
        "size" : 50000,        // we set it with total number of services exist
        "min_doc_count" : 1,
        "shard_min_doc_count" : 0,
        "show_term_doc_count_error" : false,
        "order" : [
          {
            "_count" : "desc"
          },
          {
            "_term" : "asc"
          }
        ]
      },
      "aggregations" : {
        "top" : {
          "top_hits" : {
            "from" : 0,
            "size" : 1,
            "version" : false,
            "explain" : false,
            "sort" : [
              {
                "timeChanged" : {
                  "order" : "desc"
                }
              }
            ]
          }
        }
      }
    }
  }
}

从上面的查询中，我们从滚动的第一击中获得聚合，这是聚合中最新服务状态的列表。然后通过Post过滤器，我们以10000的批量获取OPEN服务，并尝试将ID(通过Java代码)与聚合列表进行匹配，以找出我们的候选对象。

返回所需的输出花费了太多时间。索引中的4.4M记录大约需要8分钟。

如果您建议将过滤器放在返回的聚合数据上的方法，则可以解决此问题。但是在搜索了很多地方之后，我发现 flex 不支持它。是这样吗？
相同问题的引用:

Elasticsearch: filter top hits aggregation

Elasticsearch exclude top hit on field value

请帮助并提出解决方案的更好方法。

谢谢。

免责声明:请不要建议先应用查询然后进行汇总，因为它不能解决问题。例如如果我先按OPEN状态进行过滤，然后进行汇总，那么对于给定的日期，我总会获得OPEN服务，但实际上对于给定的日期，服务可能已解决。

Answer 1

这是我尝试满足您的需求。我有一个概念证明，它不能与字符串状态一起工作。因此，我们首先需要将字符串状态转换为数字(也许update by query可以为您完成工作)

在我的例子中

OPEN => status_number = 1 
CLOSED => status_number = 2 
RESOLVED => status_number = 3 

这是我的50美分要求:D

POST <index>/doc/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": {
        "range": {
          "timeChanged": {
            "gte": "2019-03-21T15:09:17.015Z",
            "lte": "2019-03-21T15:09:18.015Z"
          }
        }
      }
    }
  },
  "aggs": {
    "service": {
      "terms": {
        "field": "id.keyword",
        "size": 10
      },
      "aggs": {
        "last_status": {
          "terms": {
            "field": "status_number",
            "size": 1,
            "order": {
              "last_change": "desc" // order to keep the last status of the timespan with the size of 1
            }
          },
          "aggs": {
            "last_change": {
              "max": {
                "field": "timeChanged"
              }
            }
          }
        },
        "min_status": {
          "min_bucket": {
            "buckets_path": "last_status._key" // used to transforms a bucket list in a single value for the bucket_selector beneath
          }
        },
        "filtered": {
          "bucket_selector": {
            "buckets_path": {
              "key": ">min_status"
            },
            "script": """
              params.key == 1 // filter buckets where last status_number is 1 si status = OPEN
            """
          }
        }
      }
    }
  }
}

输出非常详细:

{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 6,
    "max_score": 0,
    "hits": []
  },
  "aggregations": {
    "service": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "servicer-id-4",
          "doc_count": 1,
          "last_status": {
            "doc_count_error_upper_bound": 0,
            "sum_other_doc_count": 0,
            "buckets": [
              {
                "key": 1,
                "doc_count": 1,
                "last_change": {
                  "value": 1553180958015,
                  "value_as_string": "2019-03-21T15:09:18.015Z"
                }
              }
            ]
          },
          "min_status": {
            "value": 1,
            "keys": [
              "1"
            ]
          }
        }
      ]
    }
  }
}

但是你只需要 aggregations.service.buckets.key值

我希望它可以为您提供帮助，但当然没有数据，我无法评估此查询的性能。