基数日期直方图

Question

为了实现表示唯一访问者指标总数的日期直方图，查询 Elasticsearch 的最佳方法是什么？

考虑以下数据：

PUT /events
{
"mappings" : {
        "_doc" : {
            "properties" : {
                "userId" : { "type" : "keyword" },
                "eventDate" : { "type" : "date" }
            }
        }
    }
}

POST /events/_bulk
{ "index" : { "_index" : "events", "_type" : "_doc", "_id" : "1" } }
{"userId": "1","eventDate": "2019-03-04T13:40:18.514Z"}
{ "index" : { "_index" : "events", "_type" : "_doc", "_id" : "2" } }
{"userId": "2","eventDate": "2019-03-04T13:46:18.514Z"}
{ "index" : { "_index" : "events", "_type" : "_doc", "_id" : "3" } }
{"userId": "3","eventDate": "2019-03-04T13:50:18.514Z"}
{ "index" : { "_index" : "events", "_type" : "_doc", "_id" : "4" } }
{"userId": "1","eventDate": "2019-03-05T13:46:18.514Z"}
{ "index" : { "_index" : "events", "_type" : "_doc", "_id" : "5" } }
{"userId": "4","eventDate": "2019-03-05T13:46:18.514Z"}

现在，如果我查询 userId 字段的基数，我会得到 4 个不同的访问者。

POST /events/_search
{
    "size": 0,
    "aggs": {
        "visitors": {
            "cardinality": {
                "field": "userId"
            }
        }
    }
}

但是，通过日期直方图分发文档，我得到的总和为 5，因为两个存储桶中都有重复的 userId。

POST /events/_search
{
    "size": 0,
    "aggs": {
        "visits_over_time": {
            "date_histogram": {
                "field": "eventDate",
                "interval": "1d"
            },
            "aggs": {
                "visitors": {
                    "cardinality": {
                        "field": "userId"
                    }
                }
            }
        }
    }
}

有没有办法过滤掉那些重复的值？实现这一目标的最佳方法是什么？

Answer 1

我们在代码中遇到了同样的问题，我们的解决方案是在 UserId 字段上使用术语聚合，在日期时间字段上使用嵌套的 Min 聚合。这为每个用户 ID 提供了一个存储桶，其中包含第一次访问的存储桶。我们在日期直方图之外进行这种聚合，然后手动映射它。

"aggs": {
    "UniqueUsers": {
      "terms": {
        "field": "userId",
        "size": 1000,
      }, "aggs": {
        "FirstSeen": {
          "min": {
            "field": "date"
          }
        }
      }
    }
  }

这对我们有用，但我相信应该有更好的实现。

Answer 2

尽管我想避免使用脚本，但脚本化指标聚合似乎是完成要求的唯一方法：

\n\n

{\n    "size": 0,\n    "aggs": {\n        "visitors": {\n            "scripted_metric": {\n                "init_script": "params._agg.dateMap = new HashMap();",\n                "map_script": "params._agg.dateMap.merge(doc.userId[0].toString(), doc.eventDate.value, (e1, e2) -> e1.isBefore(e2) ? e1 : e2);",\n                "combine_script": "return params._agg.dateMap;",\n                "reduce_script": "def dateMap = new HashMap(); for (map in params._aggs) { if (map == null) continue; for (entry in map.entrySet()) dateMap.merge(entry.key, entry.value, (e1, e2) -> e1.isBefore(e2) ? e1 : e2); } def hist = new TreeMap(); for (entry in dateMap.entrySet()) hist.merge(entry.value.toString(), 1, (a, b) -> a + 1); return hist;"\n            }\n        }\n    }\n}\n

\n\n

Init 只是创建一个空的 HashMap，Map 使用 userId 作为键填充该映射，并将最旧的 eventDate 设置为值，Combine 只是解开要传递给 Reduce 的映射：

\n\n

def dateMap = new HashMap();\nfor (map in params._aggs) {\n    if (map == null) continue;\n    for (entry in map.entrySet())\n        dateMap.merge(entry.key, entry.value, (e1, e2) -> e1.isBefore(e2) ? e1 : e2);\n}\n\ndef hist = new TreeMap();\nfor (entry in dateMap.entrySet())\n    hist.merge(entry.value.toString(), 1, (a, b) -> a + 1);\nreturn hist;\n

\n\n

在为每个集群节点执行合并代码之前，Reduce 将所有映射合并为一个（即 dateMap），保留每个用户 ID 的最旧的 eventDate。然后它计算每个 eventDate 的出现次数。

\n\n

结果是：

\n\n

"aggregations": {\n    "visitors": {\n        "value": {\n            "2019-03-04T13:40:18.514Z": 1,\n            "2019-03-04T13:46:18.514Z": 1,\n            "2019-03-04T13:50:18.514Z": 1,\n            "2019-03-05T13:46:18.514Z": 1\n        }\n    }\n}\n

\n\n

唯一缺少的部分是这些值必须分组为应用程序代码的直方图。

\n\n

注意\xc2\xb9：使用需要您自担风险，我不知道内存消耗是否会因为这些哈希映射而增加很多，也不知道它在大型数据集上的表现如何。

\n\n

注意\xc2\xb2：从 Elasticsearch 6.4 开始state，states应使用 和 代替params._agg和params._aggs。

\n