如何在 Elasticsearch 中按子聚合结果过滤命中

Question

我一直在实施基于 elasticsearch 的事件溯源解决方案。文档代表状态变化事件，由_source 上的id 字段链接。有一个从0 开始的sequence 字段，因此每个id 的最高序列是该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 字段上编写了一个 terms 聚合，以及两个 top_hits 子聚合，通过对序列进行排序并分别抓取顶部和底部的结果来获取第一个和最新的事件。

问题是我对状态所做的任何匹配都发生在聚合之前，我需要一种方法从术语聚合结果中排除所有最新事件的状态与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

Terms agg size 是这样的，所以这个查询可以在预定的批次中运行，总是抓取结果的顶部并更新链，这样它们就不会出现在下一个查询中。

Answer 1

我对此进行了深入研究并试图查看它。我认为这归结为单个聚合的局限性。无法对 top_hits 进行子聚合，因此我需要其他方法来过滤返回的结果。

我最终发现有人在做类似的事情：https://rahulsinghai.blogspot.com/2016/07/elasticsearch-pipeline-bucket-selector.html

输入组合top_hits、max 以查找max 序列每个id 和filter 聚合在同一级别，然后在filter 聚合上找到另一个max 聚合以查找@ 987654330@ 序列每个id 仅针对状态为READY 的每个结果，假设共享一个id 的所有事件在READY 状态下至少有一个事件，然后使用bucket_selector 聚合来选择基于@987654335 的相关集@ 和 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"
                    }
                }
            }
        }
    }
}