ElasticSearch如何查询最长的任务

Question

我在 Elastic Search 中有以下格式的数据:

POST slots/slot/1
{
    taskId:1,
    datetime: "2020-05-10T08:45:44",
    status: "START",
}

POST slots/slot/2
{
    taskId:1,
    datetime: "2020-05-10T08:49:54",
    status: "STOP",
}
...

并想找到一种方法来检索前 3 个运行时间最长的任务(这意味着任务，同时存在 START 和 STOP json 对象，并且它的 START/STOP 时间之间的差异是最长的) - 我想检索 taskId和 runningTime(= 任务运行了多长时间)。

是否可以在 ElasticSearch 中完成此任务？ ElasticSearch 是否适合此类任务？

请宽容，我是 ElasticSearch 技术的新手。

Answer 1

这个很棘手。假设您将为每个unique taskId 恰好 2 个文档，其中一个是START 和另一个 STOP。在这种情况下，我们可以执行以下操作:

GET slots/_search
{
  "size": 0,
  "aggs": {
    "by_ids": {
      "terms": {
        "field": "taskId",
        "size": 10000,
        "min_doc_count": 2
      },
      "aggs": {
        "start_bucket": {
          "filter": {
            "term": {
              "status.keyword": "START"
            }
          },
          "aggs": {
            "datetime_term": {
              "max": {
                "field": "datetime"
              }
            }
          }
        },
        "stop_bucket": {
          "filter": {
            "term": {
              "status.keyword": "STOP"
            }
          },
          "aggs": {
            "datetime_term": {
              "max": {
                "field": "datetime"
              }
            }
          }
        },
        "diff_in_millis": {
          "bucket_script": {
            "buckets_path": {
              "start": "start_bucket.datetime_term",
              "stop": "stop_bucket.datetime_term"
            },
            "script": "return params.stop - params.start"
          }
        },
        "final_sort": {
          "bucket_sort": {
            "sort": [
              {
                "diff_in_millis": {
                  "order": "desc"
                }
              }
            ],
            "size": 3
          }
        }
      }
    }
  }
}

根据 this discussion ,

the caveat is that this performs sorting on the final list of buckets. So if a term isn't in the list, it won't get sorted. That's in contrast to sorting on the terms agg itself, which changes the contents of the list.

换句话说，我们需要将顶级 size 设置任意高，以便我们所有的 taskIDs 得到聚合。和/或使用例如仅 2020 年或上个月等的日期过滤器来预过滤上下文，这样我们就可以减少覆盖范围并节省一些 CPU 处理时间。

如果一切顺利，并且您的 status 有一个 .keyword 字段(更多关于此 here )我们可以过滤，您最终会得到所有您需要的信息:

{
  ...
  "aggregations":{
    "by_ids":{
      "doc_count_error_upper_bound":0,
      "sum_other_doc_count":0,
      "buckets":[
        {
          "key":2,            <-- taskID (this one was added by myself)
          "doc_count":2,
          "start_bucket":{
            ...
          },
          "stop_bucket":{
            ...
          },
          "diff_in_millis":{
            "value":3850000.0        <-- duration in millis
          }
        },
        {
          "key":1,                  <-- task from the question
          "doc_count":2,
          "start_bucket":{
            ...
          },
          "stop_bucket":{
           ...
          },
          "diff_in_millis":{
            "value":250000.0        <-- duration in millis
          }
        }
      ]
    }
  }
}

编辑/更正:

"min_doc_count": 2 是必需的 b/c 我们只对实际完成的任务感兴趣。如果您想包括那些已经运行但尚未完成的任务，请创建另一个赏金任务；)