Фильтрация документов после агрегации - PullRequest
Новая
       9

Фильтрация документов после агрегации

0 голосов
/ 24 сентября 2018

В Elasticsearch я храню моментальные снимки состояния элемента в схеме только для добавления.Например: 

POST /item/item
{
  "id": "1",
  "time": "2018-09-19T00:00:00Z",
  status": "ON_HOLD"
}

POST /item/item
{
  "id": "2",
  "time": "2018-09-19T00:01:00Z",
  "status": "ON_HOLD"
}

POST /item/item
{
  "id": "2",
  "time": "2018-09-19T00:02:00Z",
  "status": "DONE"
}

Теперь я хочу ответить на следующий вопрос: какие пункты еще удерживаются? (status==ON_HOLD).

В этом простом случае ответом будет: 

{
  "id": "1",
  "time": "2018-09-19T00:00:00Z",
  status": "ON_HOLD"
}

Итак, чтобы получить последнее состояние элемента, я использую агрегирование терминов на id, например, так: 

GET /item/_search
{
  "size": 0,
  "query": {
    "match_all": {}
  },
  "aggs": {
    "id": {
      "terms": {
        "field": "id.keyword",
        "size": 10
      },
      "aggs": {
        "top_items": {
          "top_hits": {
            "size": 1,
            "sort": [
              {
                "time": {
                  "order": "desc"
                }
              }
            ],
            "_source": {
              "includes": ["*"]
            }
          }
        }
      }
    }
  }
}

Это дает мне последнее доступное состояние каждого элемента, идентифицируемого по его идентификатору: 

{
  "took": 1,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 3,
    "max_score": 0,
    "hits": []
  },
  "aggregations": {
    "id": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "2",
          "doc_count": 2,
          "top_items": {
            "hits": {
              "total": 2,
              "max_score": null,
              "hits": [
                {
                  "_index": "item",
                  "_type": "item",
                  "_id": "S-5eCGYBNyILygyml2jR",
                  "_score": null,
                  "_source": {
                    "id": "2",
                    "time": "2018-09-19T00:02:00Z",
                    "status": "DONE"
                  },
                  "sort": [
                    1537315320000
                  ]
                }
              ]
            }
          }
        },
        {
          "key": "1",
          "doc_count": 1,
          "top_items": {
            "hits": {
              "total": 1,
              "max_score": null,
              "hits": [
                {
                  "_index": "item",
                  "_type": "item",
                  "_id": "Se5eCGYBNyILygymjmg0",
                  "_score": null,
                  "_source": {
                    "id": "1",
                    "time": "2018-09-19T00:00:00Z",
                    "status": "ON_HOLD"
                  },
                  "sort": [
                    1537315200000
                  ]
                }
              ]
            }
          }
        }
      ]
    }
  }
}

Теперь проблема в том, что я хотел бы отфильтровать результат (после агрегации) на стороне Elasticsearch (не клиента)).

Я попытался агрегировать bucket_selector, но он жалуется, поскольку результат top_hits не является числовым или числовым агрегированием.

Я также пытался добавить поле script_, чтобы получитьчисловое значение, но не может использовать его после: 

"script_fields": {
  "on_hold": {
    "script": {
      "lang": "painless",
      "source": "doc['status.keyword'].value == 'ON_HOLD' ? 1 : 0"
    }
  }
}

Возможно ли то, что я хочу сделать, на стороне Elasticsearch или мне нужно делать это на стороне клиента?

PS:добавление фильтра до агрегации не дает правильного результата, так как это приведет кКоличество предметов, которые были ON_HOLD в любой момент времени.

РЕДАКТИРОВАТЬ: Хорошо, я получаю где-то с: 

GET /item/_search
{
  "size": 0,
  "query": {
    "match_all": {}
  },
  "aggs": {
    "id": {
      "terms": {
        "field": "id.keyword",
        "size": 50
      },
      "aggs": {
        "top_item": {
          "terms": {
            "size": 1,
            "field": "time",
            "order": {
              "_key": "desc"
            }
          },
          "aggs": {
            "on_hold": {
              "filter": {
                "term": {
                  "status.keyword": "ON_HOLD"
                }
              },
              "aggs": {
                "document": {
                  "top_hits": {
                    "size": 1,
                    "_source": ["*"]
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

Агрегация top_hits является метрикой, а не агрегации корзины,поэтому он не выполняет свою работу и должен использоваться последним.

Еще одна проблема: отфильтрованные корзины оставляют пустые листья: "хиты": []

Есть ли способ удалить такиеветви, оканчивающиеся пустыми листьями из результирующего дерева?Спасибо

1 Ответ

0 голосов
/ 02 октября 2018

Хорошо, я нашел полное решение проблемы, включая фильтрацию пустых веток в дереве агрегации: 

GET /item/_search
{
  "size": 0,
  "query": {
    "match_all": {}
  },
  "aggs": {
    "id": {
      "terms": {
        "field": "id.keyword",
        "size": 50
      },
      "aggs": {
        "top_item": {
          "terms": {
            "size": 1,
            "field": "time",
            "order": {
              "_key": "desc"
            }
          },
          "aggs": {
            "on_hold": {
              "filter": {
                "term": {
                  "status.keyword": "ON_HOLD"
                }
              },
              "aggs": {
                "document": {
                  "top_hits": {
                    "size": 1,
                    "_source": ["*"]
                  }
                }
              }
            },
            "remove_filtered": {
              "bucket_selector": {
                "buckets_path": {
                  "count": "on_hold._count"
                },
                "script": {
                  "source": "params.count != 0"
                }
              }
            }
          }
        },
        "remove_empty": {
          "bucket_selector": {
            "buckets_path": {
              "count": "top_item._bucket_count"
            },
            "script": "params.count != 0"
          }
        }
      }
    }
  }
}

Это дает следующий ожидаемый результат: 

{
  "took": 2,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 3,
    "max_score": 0,
    "hits": []
  },
  "aggregations": {
    "id": {
      "doc_count_error_upper_bound": 0,
      "sum_other_doc_count": 0,
      "buckets": [
        {
          "key": "1",
          "doc_count": 1,
          "top_item": {
            "doc_count_error_upper_bound": 0,
            "sum_other_doc_count": 0,
            "buckets": [
              {
                "key": 1537315200000,
                "key_as_string": "2018-09-19T00:00:00.000Z",
                "doc_count": 1,
                "on_hold": {
                  "doc_count": 1,
                  "document": {
                    "hits": {
                      "total": 1,
                      "max_score": 1,
                      "hits": [
                        {
                          "_index": "item",
                          "_type": "item",
                          "_id": "HvywM2YB5Ei0wOZMeia9",
                          "_score": 1,
                          "_source": {
                            "id": "1",
                            "time": "2018-09-19T00:00:00Z",
                            "status": "ON_HOLD"
                          }
                        }
                      ]
                    }
                  }
                }
              }
            ]
          }
        }
      ]
    }
  }
}
...