Elasticsearch - 查询以从扁平结构中获取最新版本的记录答案

【问题标题】：Elasticsearch - query to get latest version of records from a flattened structureElasticsearch - 查询以从扁平结构中获取最新版本的记录
【发布时间】：2019-10-15 11:48:51
【问题描述】：

我有一个场景，我想从 Elasticsearch 中按某个键值分组的索引中返回最新的非规范化数据 - 在下面的场景中 => TradeRef。

下图更好地描绘了保存在索引中的数据：

{"Row": "1", "TradeRef": "A", "TradeRefDate": "2019-01-01 13:00", "TradeRefId": "FFF", "MessageId": "XXX", "MessageStatus": "S-Open"}, 
{"Row": "2", "TradeRef": "B", "TradeRefDate": "2019-01-01 13:00", "TradeRefId": "GGG", "MessageId": "YYY", "MessageStatus": "P-Open"},
{"Row": "3", "TradeRef": "C", "TradeRefDate": "2019-01-01 13:00", "TradeRefId": "HHH", "MessageId": "ZZZ", "MessageStatus": "Q-Open"},
{"Row": "4", "TradeRef": "A", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "III", "MessageId": "AAA", "MessageStatus": "R-Open"},
{"Row": "5", "TradeRef": "B", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "JJJ", "MessageId": "BBB", "MessageStatus": "T-Open"},
{"Row": "6", "TradeRef": "A", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "III", "MessageId": "CCC", "MessageStatus": "R-Open"},
{"Row": "7", "TradeRef": "B", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "JJJ", "MessageId": "DDD", "MessageStatus": "T-Open"}

我希望我的查询返回以下结果，其中第 1 行和第 2 行被删除，因为它们引用了具有较旧 TradeRefDate (2019-01-01 13:00) 的 Trade Refs 'A' 和 'B'。

索引中更新的行包含相同的 TradeRef 'A' 和 'B' 以及更新的 TradeRefDate (2019-01-01 14:00)：

{"Row": "3", "TradeRef": "C", "TradeRefDate": "2019-01-01 13:00", "TradeRefId": "HHH", "MessageId": "ZZZ", "MessageStatus": "Q-Open"},
{"Row": "4", "TradeRef": "A", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "III", "MessageId": "AAA", "MessageStatus": "R-Open"},
{"Row": "5", "TradeRef": "B", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "JJJ", "MessageId": "BBB", "MessageStatus": "T-Open"},
{"Row": "6", "TradeRef": "A", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "III", "MessageId": "CCC", "MessageStatus": "R-Open"},
{"Row": "7", "TradeRef": "B", "TradeRefDate": "2019-01-01 14:00", "TradeRefId": "JJJ", "MessageId": "DDD", "MessageStatus": "T-Open"}

我们将不胜感激。我尝试了以下查询，但它只为每个 TradeRef 提供一行，而不是与最新的 TradeRef 值关联的匹配记录：

GET /flattened_index_v1/_search
{
  "from": 0,
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "TradeRefDate": {
              "gte": "2018-09-01T00:00:00",
              "lte": "2019-10-26T00:00:00"
            }
          }
        },
        {
          "exists": {
            "field": "MessageId"
          }
        }
      ]
    }
  },
  "size": 0,
  "aggs": {
    "grp_by_trade_ref": {
      "terms": {
        "field": "TradeRef.keyword",
        "size": 1000
      },
      "aggs": {
        "latest_trecs": {
          "top_hits": {
            "size": 1,
            "sort": [
              {
                "TradeRefDate": {
                  "order": "desc"
                }
              }
            ],
            "_source": {"includes": ["TradeRef", "TradeRefId", "MessageId", "MessageStatus", "TradeRefDate"]}
          }
        }
      }
    }
  }
}

【问题讨论】：

标签： elasticsearch elasticsearch-aggregation

【解决方案1】：

对关键字进行术语聚合
对关键字下的日期进行术语聚合。 i 根据日期降序选择前 1 ii.返回 top_hits

GET index22/_search
{
  "size": 0,
  "aggs": {
    "TradeRef": {
      "terms": {
        "field": "TradeRef.keyword",
        "size": 10
      },
      "aggs": {
        "RefDate": {
          "terms": {
            "field": "TradeRefDate",
            "order": {
              "_term": "desc"
            },
            "size": 1
          },
          "aggs": {
            "TopDocuments": {
              "top_hits": {
                "size": 10
              }
            }
          }
        }
      }
    }
  }
}

编辑 1：你可以使用（复合聚合）[https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-composite-aggregation.html]

在复合聚合中，您可以使用 after_key 连续分页，即您可以先获取 n 条记录，然后再获取下 n 条记录，您不能从第 1 页跳转到第 3 页。

GET index22/_search
{
  "size": 0,
  "aggs": {
    "pagination": {
      "composite": {
       "size": 2,  ---> page_size
        "sources": [
          {
            "TradeRef": {
              "terms": {
                "field": "TradeRef.keyword"
              }
            }
          }
        ]
      },
      "aggs": {
        "RefDate": {
          "terms": {
            "field": "TradeRefDate",
            "order": {
              "_term": "desc"
            },
            "size": 1
          },
          "aggs": {
            "TopDocuments": {
              "top_hits": {
                "size": 10
              }
            }
          }
        }
      }
    }
  }
}

回复：

"aggregations" : {
    "pagination" : {
      "after_key" : {
        "TradeRef" : "B"    ----> use to fetch next set of records.
      },
      "buckets" : [
        {
          "key" : {
            "TradeRef" : "A"
          },
          "doc_count" : 3,
          "RefDate" : {
            "doc_count_error_upper_bound" : 0,
            "sum_other_doc_count" : 1,
            "buckets" : [
              {
                "key" : 50460000,
                "key_as_string" : "1970-01-01 14:00",
                "doc_count" : 2,
                "TopDocuments" : {
                  "hits" : {
                    "total" : {
                      "value" : 2,
                      "relation" : "eq"
                    },
                    "max_score" : 1.0,
                    "hits" : [
                      {
                        "_index" : "index22",
                        "_type" : "_doc",
                        "_id" : "IkBnyG0BwSpwFwW4UeB7",
                        "_score" : 1.0,
                        "_source" : {
                          "Row" : "4",
                          "TradeRef" : "A",
                          "TradeRefDate" : "2019-01-01 14:00",
                          "TradeRefId" : "III",
                          "MessageId" : "AAA",
                          "MessageStatus" : "R-Open"
                        }
                      },
                      {
                        "_index" : "index22",
                        "_type" : "_doc",
                        "_id" : "JEBnyG0BwSpwFwW4ceBs",
                        "_score" : 1.0,
                        "_source" : {
                          "Row" : "6",
                          "TradeRef" : "A",
                          "TradeRefDate" : "2019-01-01 14:00",
                          "TradeRefId" : "III",
                          "MessageId" : "CCC",
                          "MessageStatus" : "R-Open"
                        }
                      }
                    ]
                  }
                }
              }
            ]
          }
        },
        {
          "key" : {
            "TradeRef" : "B"
          },
          "doc_count" : 3,
          "RefDate" : {
            "doc_count_error_upper_bound" : 0,
            "sum_other_doc_count" : 1,
            "buckets" : [
              {
                "key" : 50460000,
                "key_as_string" : "1970-01-01 14:00",
                "doc_count" : 2,
                "TopDocuments" : {
                  "hits" : {
                    "total" : {
                      "value" : 2,
                      "relation" : "eq"
                    },
                    "max_score" : 1.0,
                    "hits" : [
                      {
                        "_index" : "index22",
                        "_type" : "_doc",
                        "_id" : "I0BnyG0BwSpwFwW4V-DW",
                        "_score" : 1.0,
                        "_source" : {
                          "Row" : "5",
                          "TradeRef" : "B",
                          "TradeRefDate" : "2019-01-01 14:00",
                          "TradeRefId" : "JJJ",
                          "MessageId" : "BBB",
                          "MessageStatus" : "T-Open"
                        }
                      },
                      {
                        "_index" : "index22",
                        "_type" : "_doc",
                        "_id" : "JUBnyG0BwSpwFwW4h-Cq",
                        "_score" : 1.0,
                        "_source" : {
                          "Row" : "7",
                          "TradeRef" : "B",
                          "TradeRefDate" : "2019-01-01 14:00",
                          "TradeRefId" : "JJJ",
                          "MessageId" : "DDD",
                          "MessageStatus" : "T-Open"
                        }
                      }
                    ]
                  }
                }
              }
            ]
          }
        }
      ]
    }
  }

【讨论】：

哇...成功了！你是明星 :) 我现在面临的下一个挑战是如何对从该查询返回的数千条记录进行分页和排序。对此的任何帮助将不胜感激。我知道我可以使用“分区聚合”将结果分解为记录块，但我知道这可能会对内存性能产生挑战。
感谢 Jaspreet 的额外更新。您提供的查询确实通过传递给 size 参数的值来限制 TradeRef 记录，但不限制每个 TradeRef 中包含的命中。我猜这是因为 ES 在桶中返回结果的方式？只是想知道是否有一种方法可以将 TradeRef 中的所有命中组合到一个存储桶中，并对组合的结果集应用排序和分页？
Top hits 支持 from、size 和 sort as in query 你可以用它来分页 top hits
复合聚合不适用于子聚合。所以我们不能从所有 sub termsl 中创建一个桶。您的目标可以使用复合聚合+ top_hit from 和 size 来实现。
谢谢贾斯普雷特。我面临的挑战是 top_hits 排序和大小仅适用于每个存储桶级别包含的值，在这种情况下，这些值已按 TradeRef 分组。如果我能够将每个存储桶中包含的值组合到一个新的“存储桶”中，我可以在其中按组合记录中的任何字段进行排序，那对我来说将是可取的。能够使用“after_key”获取下一批记录非常有帮助 - 再次感谢。