elasticsearch - 使用 Elasticsearch 进行总和/平均的子聚合

Question

我有一个包含 user_id、amount 和 category 字段的索引交易。
我想计算每个用户和类别的平均金额，然后以获取每个类别的总平均金额结束。
SQL 看起来像这样:

SELET AVG(average), category from

    (SELECT user_id, category, AVG(amount) AS average FROM transactions WHERE amount < 100000 
    GROUP BY user_id, category) AS a1

GROUP BY category

我只是得到一个包含所有用户 ID 的存储桶的响应，然后在其中包含每个类别(用户)的平均金额的存储桶。我不明白如何添加另一个聚合来完成我所追求的。

{
  "aggs": {
    "group_by_users": {
      "terms": {
        "field": "user_id.keyword"
      },
      "aggs": {
        "group_by_category": {
          "terms": {
              "field": "category.keyword"
          },
          "aggs": {
            "average_amount": {
              "avg": {
                "field": "amount"
              }
            }
          }
        }
      }
    }
  }
}

很感谢任何形式的帮助。

编辑:请求示例，因此这里首先是一些示例数据，然后是中间结果，该结果将以底部的所需结果结束。

-----------------------------------------
|  user_id  |   category   |   amount   |
-----------------------------------------
|     1     |   insurances |   1000     |
|     1     |   transport  |     50     |
|     1     |   transport  |    100     |
|     2     |   insurances |    700     |
|     2     |   insurances |    200     |
|     2     |   transport  |    300     |
-----------------------------------------

用户1传输计算:(50+100)/2

因此，需要发生的第一件事是按 user_id 和类别分组以获得每个用户和类别的平均值。

这将产生:

-----------------------------------------
|  user_id  |   category   |   average  |
-----------------------------------------
|     1     |   insurances |   1000     |
|     1     |   transport  |     75     |
|     2     |   insurances |    450     |
|     2     |   transport  |    300     |
-----------------------------------------

重要的是要明白我不能为所有用户做一个平均值，我首先需要每个用户、每个类别的平均支出。

所以现在我只想按类别分组并计算平均金额:

-----------------------------
|   category   |   average  |
-----------------------------
|   insurances |   725      |
|   transport  |   187,5    |
-----------------------------

保险示例:(1000 + 450)/2

Answer 1

avg_bucket pipeline aggregation在某些情况下会完成这项工作(但可能无法随着索引的大小很好地扩展，请参阅下面的注释):

POST myindex1/_search
{
  "size": 0,
  "aggs": {
    "by category": {
      "terms": {
        "field": "category.keyword"
      },
      "aggs": {
        "by user_id": {
          "terms": {
            "field": "user_id"
          },
          "aggs": {
            "avg by user": {
              "avg": {
                "field": "amount"
              }
            }
          }
        },
        "average by user, category": {
          "avg_bucket": {
            "buckets_path": "by user_id>avg by user"
          }
        }
      }
    }
  }
}

这将响应如下所示:

{
  ...
  "aggregations" : {
    "by category" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "insurances",
          "doc_count" : 3,
          "by user_id" : {
            "doc_count_error_upper_bound" : 0,
            "sum_other_doc_count" : 0,
            "buckets" : [
              {
                "key" : 2,
                "doc_count" : 2,
                "avg by user" : {
                  "value" : 450.0
                }
              },
              {
                "key" : 1,
                "doc_count" : 1,
                "avg by user" : {
                  "value" : 1000.0
                }
              }
            ]
          },
          "average by user, category" : {
            "value" : 725.0   <--- average for `insurances`
          }
        },
        {
          "key" : "transport",
          "doc_count" : 3,
          "by user_id" : {
            "doc_count_error_upper_bound" : 0,
            "sum_other_doc_count" : 0,
            "buckets" : [
              {
                "key" : 1,
                "doc_count" : 2,
                "avg by user" : {
                  "value" : 75.0
                }
              },
              {
                "key" : 2,
                "doc_count" : 1,
                "avg by user" : {
                  "value" : 300.0
                }
              }
            ]
          },
          "average by user, category" : {
            "value" : 187.5      <--- average for `transport`
          }
        }
      ]
    }
  }
}

它是如何工作的？

让我们从 "by user_id" 开始 terms 聚合:我们要求 Elasticsearch 按 user_id 对文档进行分组并计算 amount 的平均值使用 avg 聚合。

POST myindex1/_search
{
  "size": 0,
  "aggs": {
    "by user_id": {
      "terms": {
        "field": "user_id"
      },
      "aggs": {
        "avg by user": {
          "avg": {
            "field": "amount"
          }
        }
      }
    }
  }
}

这相当于 SQL:

SELECT user_id, avg(amount)
FROM my_index
GROUP BY user_id;

到目前为止，这非常简单。但是我们现在如何对这个类别进行平均呢？

我们可以再添加一个 terms聚合 "by category"顶部 "by user id"一。现在 avg还将考虑以下类别:

POST myindex1/_search
{
  "size": 0,
  "aggs": {
    "by category": {
      "terms": {
        "field": "category.keyword"
      },
      "aggs": {
        "by user_id": {
          "terms": {
            "field": "user_id"
          },
          "aggs": {
            "avg by user": {
              "avg": {
                "field": "amount"
              }
            }
          }
        }
      }
    }
  }
}

这相当于 SQL:

SELECT user_id, category, avg(amount)
FROM my_index
GROUP BY user_id, category;

我们现在可以使用上一个查询的结果，并在 category 上聚合吗？再次？

这可以通过 avg_bucket 来完成管道聚合。唯一缺少的是告诉 avg_bucket聚合在哪里确切地找到要聚合的桶，这是通过 buckets_path 完成的表达。

这就是我们如何到达我在顶部发布的查询，它将有效地执行您在问题中发布的 SQL。

但...

会出什么问题？

这种方法的缺点是它不能很好地随着索引中的文档数量而扩展。

事实是， pipeline aggregation仅对已经聚合的数据进行操作:

Pipeline aggregations work on the outputs produced from other aggregations rather than from document sets, adding information to the output tree.

在我们的例子中，这意味着如果有超过 10 个不同的 user_id在指数中，我们的平均值不会很精确。

发生这种情况是因为默认情况下 terms聚合仅返回 top 10 buckets , 和 SQL-ish 等价物应该如下所示:

SELECT category, avg(avg_amount)
FROM (
    SELECT user_id, category, avg(amount) avg_amount
    FROM my_index
    GROUP BY user_id, category
    LIMIT 10 per user_id
) Q
LIMIT 10;

可以通过 size 更改此限制 terms 的参数聚合。

要记住的另一件事是 terms返回 approximate document counts ，这也会影响平均值。

希望有帮助!