elasticsearch - 使用嵌套和根级别数据的Elasticsearch嵌套聚合比率

Question

我有一个简单的聚合感觉。我有一个时序代码如下的文档:

{
  "task_start": "2020-06-03T21:19:07.908821Z",
  "task_end": "2020-06-03T21:27:00.323790Z",
  "sub_tasks": [
    {
      "key": "sub-task1-time-milliseconds",
      "value": 3310
    },
    {
      "key": "sub-task2-time-milliseconds",
      "value": 2410
    },
    ...
  ]
}

嵌套sub_tasks的位置。我想要得到的是每个子任务中的时间与整个任务时间的中值比率。整个任务时间仅为end_time - start_time。我知道如何分别汇总子任务时间和总任务时间。但我想汇总每个文档的比率。

问题在于嵌套聚合，我只能访问带有嵌套数据的数据，而在反向嵌套聚合中，我只能访问根级别的数据，而不能同时访问两者。我知道有一种方法可以使用copy_to，因此我在嵌套路径中拥有任务时间，但是我没有能力修改索引结构，也不希望有额外的存储空间。

这是我尝试过的。对于嵌套聚合:

{
  "aggs": {
    "task_metrics": {
      "nested": {
        "path": "sub_tasks"
      },
      "aggs": {
        "sub_task_metrics": {
          "filter": {
            "term": {
              "sub_tasks.key": "sub-task1-time-milliseconds"
            }
          },
          "aggs": {
            "median_time": {
              "percentiles": {
                "script": {
                  "lang": "painless",
                  "source": """
                            double task_time = (doc['task_end'].value.millis - doc['task_start'].value.millis);
                            return doc['sub_tasks.value'].value / task_time; 
                            """
                },
                "percents": 50
              }
            }
          }
        }
      }
    }
  }
}

但是在该聚合中，doc['task_start']和doc['task_end']仅返回零，因为我无权访问它们。为了获得访问权限，我还尝试了添加另一个子聚合的reverse_nested。这使我可以访问doc['task_start']和doc['task_end']，但是doc['sub_tasks.value'].value仅返回0。

似乎这应该是可能的，但是当我阅读管道聚合和其他脚本聚合时，我不相信任何一个都能满足我的要求。非常感谢您的帮助，谢谢!

Answer 1

这很棘手-已经讨论过here。

我认为您将不得不诉诸scripted_metric s和一些方法模拟，因为公开的无痛API是somewhat limited:

{
  "size": 0, 
  "aggs": {
    "task_metrics_median": {
      "scripted_metric": {
        "init_script": "state.ratios = new ArrayList();",

        "map_script": """
          // access the source incl. the nested subtasks
          def d = params._source;

          for (def subtask : d.sub_tasks) {
            // mimicking a `term` query
            if (subtask.key != 'sub-task1-time-milliseconds') break;

            // incoming as strings so parse
            def millis_end = ZonedDateTime.parse(d.task_end).toInstant().toEpochMilli();
            def millis_start = ZonedDateTime.parse(d.task_start).toInstant().toEpochMilli();

            double task_time = (millis_end - millis_start);

            // prevent zero division
            if (task_time <= 0) break;

            state['ratios'].add(subtask.value / task_time);  
          }
        """,

        "combine_script": """
            def ratios = state.ratios;
            Collections.sort(ratios);

            // trivial median calc
            double median;
            if (ratios.length % 2 == 0) {
                median = ((double)ratios[ratios.length/2] + (double)ratios[ratios.length/2 - 1])/2;
            } else {
                median = (double) ratios[ratios.length/2];
            }

            return median
        """,

        "reduce_script": "return states"
      }
    }
  }
}