使用嵌套和根级别数据的 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_metrics 和一些方法模拟，因为暴露的无痛 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"
      }
    }
  }
}