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

Question

我有一个索引 transactions，其中包含 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 对文档进行分组，并使用 avg 聚合计算 amount 的平均值。

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;

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

我们可以在"by user id" 之上添加另一个terms 聚合"by category"。现在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 只对已经聚合的数据进行操作：

管道聚合处理从其他聚合而不是文档集产生的输出，将信息添加到输出树。

在我们的例子中，这意味着如果索引中有超过 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;

这个限制可以通过terms聚合的size参数来改变。

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

---

希望有帮助！