Как отфильтровать попадания по субагрегированным результатам в Elasticsearch

Question

Я внедряю решение для поиска событий на основе упругого поиска. Документы представляют события изменения состояния, связанные полем id в источнике _. Поле sequence начинается с 0, так что самая высокая последовательность для id является последним событием для id. На практике дополнительные данные будут доступны только по первому событию, а последующие события будут содержать только те поля, которые изменились. Цель состояла в том, чтобы иметь индекс, к которому я никогда не должен отправлять обновления, только вставки.

Попытка создать запрос, который будет возвращать первое и последнее события, сгруппированные по их идентификатору, если только, если их последнее событие status соответствует READY.

Пример данных:

[  
    {  
        "_index":"events",
        "_type":"event",
        "_id":"AWcFf2N-IqNGd75vWMgc",
        "_score":1,
        "_source":{  
            "id":"event_chain-1",
            "status":"SENT",
            "sequence":1,
            "timestamp":"1541985493824",
            "export_batch_id":"103709fe-959f-4b4e-8255-ef59f18a3cf6"
        }
    },
    {  
        "_index":"events",
        "_type":"event",
        "_id":"AWbQomwoIqNGd75vWMf6",
        "_score":1,
        "_source":{  
            "id":"event_chain-1",
            "status":"READY",
            "sequence":"0",
            "timestamp":"2018-10-31T00:00:00Z"
        }
    },
    {  
        "_index":"events",
        "_type":"event",
        "_id":"AWbQomwoIqNGd75vWabc",
        "_score":1,
        "_source":{  
            "id":"event_chain-2",
            "status":"READY",
            "sequence":"0",
            "timestamp":"2018-10-31T00:00:00Z"
        }
    }
]

Я написал агрегацию терминов в поле id.keyword и две субагрегации top_hits, чтобы получить первое и последнее события, упорядочив последовательность и взяв верхний и нижний результаты соответственно.

Проблема заключается в том, что любое сопоставление, которое я выполняю с состоянием, происходит до агрегации, и мне нужен способ исключить из результатов агрегирования терминов любые попадания, в которых статус last_event не соответствует READY.

Что у меня так далеко:

POST /events/_search
{
    "size": 0,
    "query": {
        "bool": {
            "must": {
                "match": {
                    "status": "READY"
                }
            }
        }
    },
    "aggs": {
        "group_by_id": {
            "terms": {
                "field": "id.keyword",
                "order": {
                    "_term": "asc"
                },
                "size": 100
            },
            "aggs": {
                "latest_event": {
                    "top_hits": {
                        "sort": [
                            {
                                "sequence": {
                                    "order": "desc"
                                }
                            }
                        ],
                        "from": 0,
                        "size": 1
                    }
                },
                "first_event": {
                    "top_hits": {
                        "sort": [
                            {
                                "sequence": {
                                    "order": "asc"
                                }
                            }
                        ],
                        "from": 0,
                        "size": 1
                    }
                }
            }
        },
        "num_ready": {
            "cardinality": {
                "field": "id.keyword"
            }
        }
    }
}

Это вернет два термина, один для event_chain-1 и один для event_chain-2, когда мне нужен только один для event_chain-2

Термины agg size, поэтому этот запрос можно запускать запланированными партиями, всегда очищая верхнюю часть результатов и обновляя цепочки, чтобы они не появлялись в следующем запросе.

Answer 1

Я углубился в это и попытался взглянуть на это. Я думаю, что это сводилось к ограничениям отдельных агрегатов. Не могу выполнить подагг на top_hits, поэтому мне нужен был какой-то другой способ отфильтровать полученные результаты.

В конце концов я обнаружил, что кто-то делает что-то подобное: https://rahulsinghai.blogspot.com/2016/07/elasticsearch-pipeline-bucket-selector.html

Введите объединение top_hits, max, чтобы найти последовательность max для агрегаций id и filter на том же уровне, затем еще одну агрегацию max в агрегациях filter, чтобы найти max последовательность на id только для каждого результата, находящегося в состоянии READY, при условии, что все события, имеющие общий идентификатор, имеют как минимум одно событие в состоянии READY, затем с помощью агрегации bucket_selector для выбора соответствующего набора на основе max и filter результаты.

Потенциальное решение:

POST /events/_search
{
    "size": 0,
    "aggs": {
        "grouped_by_id": {
            "terms": {
                "field": "id.keyword",
                "size": 100,
                "order": {"max_seq":"desc"}
            },
            "aggs": {
                "max_seq": {"max":{"field":"sequence"}},
                "latest_event": {
                    "top_hits": {
                        "sort": [{"sequence":{"order":"desc"}}],
                        "from": 0,
                        "size": 1
                    }
                },
                "first_event": {
                    "top_hits": {
                        "sort": [{"sequence":{"order":"asc"}}],
                        "from": 0,
                        "size": 1
                    }
                },
                "filters": {
                    "filter": {"bool":{"must":[{"match":{"status":"READY"}}]}},
                    "aggs": {
                        "latest_ready_seq": {"max":{"field":"sequence"}}
                    }
                },
                "should_we_consider": {
                    "bucket_selector": {
                        "buckets_path": {
                            "latest_seq": "max_seq",
                            "latest_ready_seq": "filters>latest_ready_seq"
                        },
                        "script": "params.latest_seq == params.latest_ready_seq"
                    }
                }
            }
        }
    }
}