博客 / 工程

在 ClickHouse 中使用聚合组合器

Denys Golotiuk

2023 年 2 月 8 日 - 15 分钟阅读

Aggregate combinators v02.png

ClickHouse 不仅支持标准聚合函数，还支持许多更高级的聚合函数，以涵盖大多数分析用例。除了聚合函数，ClickHouse 还提供聚合组合器，这是查询功能的强大扩展，可以满足大量需求。

组合器允许扩展和混合聚合，以处理各种数据结构。此功能使我们能够调整查询而不是表，以回答即使是最复杂的问题。

在这篇博文中，我们将探讨聚合组合器以及它们如何潜在地简化您的查询，并避免对数据进行结构更改的需求。

要使用组合器，我们必须做两件事。首先，选择我们要使用的聚合函数；假设我们想要一个 sum() 函数。其次，选择我们需要的组合器；假设我们需要一个 If 组合器。要在查询中使用它，我们将组合器添加到函数名称中


SELECT sumIf(...)

一个更有用的功能是，我们可以在单个函数中组合任意数量的组合器


SELECT sumArrayIf(...)

在这里，我们将 sum() 函数与 Array 和 If 组合器结合使用

这个特定的示例将允许我们有条件地对数组列的内容求和。

让我们探讨一些可以使用组合器的实际案例。

有时，我们需要根据特定条件聚合数据。我们可以使用 If 组合器，并将条件指定为组合函数的最后一个参数，而不是为此使用 WHERE 子句

how sumIf works

假设我们有一个包含用户付款的表，结构如下（使用示例数据填充）


CREATE TABLE payments
(
    `total_amount` Float,
    `status` ENUM('declined', 'confirmed'),
    `create_time` DateTime,
    `confirm_time` DateTime
)
ENGINE = MergeTree
ORDER BY (status, create_time)

假设我们想要获得支出的总金额，但仅当付款已确认时，即 status="confirmed"


SELECT sumIf(total_amount, status = 'confirmed') FROM payments

┌─sumIf(total_amount, equals(status, 'declined'))─┐
│                               10780.18000793457 │
└─────────────────────────────────────────────────┘

我们可以对条件使用与 WHERE 子句相同的语法。让我们获取已确认付款的总金额，但当 confirm_time 比 create_time 晚 1 分钟时


SELECT sumIf(total_amount, (status = 'confirmed') AND (confirm_time > (create_time + toIntervalMinute(1)))) AS confirmed_and_checked
FROM payments

┌─confirmed_and_checked─┐
│     11195.98991394043 │
└───────────────────────┘

与标准 WHERE 子句相比，使用条件 If 的主要优势在于能够为不同的子句计算多个总和。我们还可以将任何可用的聚合函数与组合器一起使用，例如 countIf()、avgIf() 或 quantileIf() - 任何函数。结合这些功能，我们可以在单个请求中对多个条件和函数进行聚合


SELECT
    countIf((status = 'confirmed') AND (confirm_time > (create_time + toIntervalMinute(1)))) AS num_confirmed_checked,
    sumIf(total_amount, (status = 'confirmed') AND (confirm_time > (create_time + toIntervalMinute(1)))) AS confirmed_checked_amount,
    countIf(status = 'declined') AS num_declined,
    sumIf(total_amount, status = 'declined') AS dec_amount,
    avgIf(total_amount, status = 'declined') AS dec_average
FROM payments

┌─num_confirmed_checked─┬─confirmed_checked_amount─┬─num_declined─┬────────dec_amount─┬───────dec_average─┐
│                    39 │        11195.98991394043 │           50 │ 10780.18000793457 │ 215.6036001586914 │
└───────────────────────┴──────────────────────────┴──────────────┴───────────────────┴───────────────────┘

计算唯一条目的数量是一种常见情况。ClickHouse 有几种方法可以使用 COUNT(DISTINCT col) （与 uniqExact 相同）或 uniq() （当估计值（但更快）足够时）来执行此操作。但是，我们可能希望从不同聚合函数中使用的列中获取唯一值。Distinct 组合器可用于此目的

Distinct combinator

一旦我们将 Distinct 添加到聚合函数，它将忽略重复值


SELECT
    countDistinct(toHour(create_time)) AS hours,
    avgDistinct(toHour(create_time)) AS avg_hour,
    avg(toHour(create_time)) AS avg_hour_all
FROM payments

┌─hours─┬─avg_hour─┬─avg_hour_all─┐
│     2 │     13.5 │        13.74 │
└───────┴──────────┴──────────────┘

在这里，avg_hour 将仅基于两个不同的值计算，而 avg_hour_all 将基于表中的所有 100 条记录计算。

由于组合器可以组合在一起，我们可以将之前的两个组合器与 avgDistinctIf 函数一起使用，以处理更高级的逻辑


SELECT avgDistinctIf(toHour(create_time), total_amount > 400) AS avg_hour
FROM payments

┌─avg_hour─┐
│       13 │
└──────────┘

这将计算 total_amount 值大于 400 的记录的不同小时值的平均值。

我们可能希望将数据拆分为组并分别计算每个组的数字，而不是进行最小/最大分析。这可以使用 Resample 组合器来解决。

它接受一个列、范围（开始/停止）和您想要拆分数据的步长。然后，它返回每个组的聚合值

Resample combinator

假设我们想要根据 total_amount 将我们的 payments 表数据从 0 （最小值）拆分到 500 （最大值），步长为 100。然后，我们想知道每个组中有多少条目以及组的平均总数


SELECT
    countResample(0, 500, 100)(toInt16(total_amount)) AS group_entries,
    avgResample(0, 500, 100)(total_amount, toInt16(total_amount)) AS group_totals
FROM payments
FORMAT Vertical

Row 1:
──────
group_entries: [21,20,24,31,4]
group_totals:  [50.21238123802912,157.32600135803222,246.1433334350586,356.2583834740423,415.2425003051758]

在这里，countResample() 函数计算每个组中的条目数，而 avgResample() 函数计算每个组的 total_amount 的平均值。Resample 组合器接受要基于其拆分的列名作为组合函数的最后一个参数。

请注意，countResample() 函数只有一个参数（因为 count() 根本不需要参数），而 avgResample() 有两个参数（第一个参数是用于计算平均值的列）。最后，我们必须使用 toInt16 将 total_amount 转换为整数，因为 Resample 组合器需要这样做。

要以表格布局获取 Resample() 组合器的输出，我们可以使用 arrayZip() 和 arrayJoin() 函数


SELECT
    round(tp.2, 2) AS avg_total,
    tp.1 AS entries
FROM
(
    SELECT
      arrayJoin(arrayZip(countResample(0, 500, 100)(toInt16(total_amount)),
      avgResample(0, 500, 100)(total_amount, toInt16(total_amount)))) AS tp
    FROM payments
)

┌─avg_total─┬─entries─┐
│     50.21 │      21 │
│    157.33 │      20 │
│    246.14 │      24 │
│    356.26 │      31 │
│    415.24 │       4 │
└───────────┴─────────┘

在这里，我们将来自 2 个数组的对应值组合成元组，并使用 arrayJoin() 函数将结果数组展开为表

当结果集不包含数据时，聚合函数的反应不同。例如，count() 将返回 0，而 avg() 将生成 nan 值。

我们可以使用 OrDefault() 和 OrNull() 组合器来控制此行为。两者都会更改在空数据集情况下使用的聚合函数的返回值

OrDefault() 将返回函数的默认值而不是 nan，
OrNull() 将返回 NULL （并且还将返回类型更改为 Nullable）。

考虑以下示例


SELECT
    count(),
    countOrNull(),
    avg(total_amount),
    avgOrDefault(total_amount),
    sumOrNull(total_amount)
FROM payments
WHERE total_amount > 1000

┌─count()─┬─countOrNull()─┬─avg(total_amount)─┬─avgOrDefault(total_amount)─┬─sumOrNull(total_amount)─┐
│       0 │          ᴺᵁᴸᴸ │               nan │                          0 │                    ᴺᵁᴸᴸ │
└─────────┴───────────────┴───────────────────┴────────────────────────────┴─────────────────────────┘

正如我们在第一列中看到的，返回了零行。请注意 countOrNull() 将返回 NULL 而不是 0，并且 avgOrDefault() 给出 0 而不是 nan。

与所有其他组合器一样，orNull() 和 orDefault() 可以与不同的组合器一起使用，以获得更高级的逻辑


SELECT
    sumIfOrNull(total_amount, status = 'declined') AS declined,
    countIfDistinctOrNull(total_amount, status = 'confirmed') AS confirmed_distinct
FROM payments
WHERE total_amount > 420

┌─declined─┬─confirmed_distinct─┐
│     ᴺᵁᴸᴸ │                  1 │
└──────────┴────────────────────┘

我们使用了 sumIfOrNull() 组合函数来仅计算已拒绝的付款，并在空集上返回 NULL。countIfDistinctOrNull() 函数计算不同的 total_amount 值，但仅针对满足指定条件的行。

ClickHouse 的 Array 类型在其用户中很受欢迎，因为它为表结构带来了很大的灵活性。为了有效地处理 Array 列，ClickHouse 提供了一组数组函数。为了简化 Array 类型上的聚合，ClickHouse 提供了 Array() 组合器。这些组合器将给定的聚合函数应用于数组列中的所有值，而不是数组本身

Array combinator

假设我们有下表（使用示例数据填充）


CREATE TABLE article_reads
(
    `time` DateTime,
    `article_id` UInt32,
    `sections` Array(UInt16),
    `times` Array(UInt16),
    `user_id` UInt32
)
ENGINE = MergeTree
ORDER BY (article_id, time)

┌────────────────time─┬─article_id─┬─sections─────────────────────┬─times────────────────────────────────┬─user_id─┐
│ 2023-01-18 23:44:17 │         10 │ [16,18,7,21,23,22,11,19,9,8] │ [82,96,294,253,292,66,44,256,222,86] │     424 │
│ 2023-01-20 22:53:00 │         10 │ [21,8]                       │ [30,176]                             │     271 │
│ 2023-01-21 03:05:19 │         10 │ [24,11,23,9]                 │ [178,177,172,105]                    │     536 │
...

此表用于存储文章每个部分的阅读数据。当用户阅读文章时，我们将阅读的部分保存到 sections 数组列，并将相关的阅读时间保存到 times 列

让我们使用 uniqArray() 函数来计算每篇文章阅读的唯一部分数量，并使用 avgArray() 获取每个部分的平均时间


SELECT
    article_id,
    uniqArray(sections) sections_read,
    round(avgArray(times)) time_per_section
FROM article_reads
GROUP BY article_id

┌─article_id─┬─sections_read─┬─time_per_section─┐
│         14 │            22 │              175 │
│         18 │            25 │              159 │
...
│         17 │            25 │              170 │
└────────────┴───────────────┴──────────────────┘

我们可以使用 minArray() 和 maxArray() 函数获取所有文章中的最小和最大阅读时间


SELECT
    minArray(times),
    maxArray(times)
FROM article_reads

┌─minArray(times)─┬─maxArray(times)─┐
│              30 │             300 │
└─────────────────┴─────────────────┘

我们还可以使用 groupUniqArray() 函数与 Array() 组合器结合使用，获取每篇文章的阅读部分列表


SELECT
    article_id,
    groupUniqArrayArray(sections)
FROM article_reads
GROUP BY article_id

┌─article_id─┬─groupUniqArrayArray(sections)───────────────────────────────────────┐
│         14 │ [16,13,24,8,10,3,9,19,23,14,7,25,2,1,21,18,12,17,22,4,6,5]          │
...
│         17 │ [16,11,13,8,24,10,3,9,23,19,14,7,25,20,2,1,15,21,6,5,12,22,4,17,18] │
└────────────┴─────────────────────────────────────────────────────────────────────┘

另一个常用的函数是 any()，它返回聚合下的任何列值，也可以与 Array 组合


SELECT
    article_id,
    anyArray(sections)
FROM article_reads
GROUP BY article_id

┌─article_id─┬─anyArray(sections)─┐
│         14 │                 19 │
│         18 │                  6 │
│         19 │                 25 │
│         15 │                 15 │
│         20 │                  1 │
│         16 │                 23 │
│         12 │                 16 │
│         11 │                  2 │
│         10 │                 16 │
│         13 │                  9 │
│         17 │                 20 │
└────────────┴────────────────────┘

Array 组合器可以与任何其他组合器一起使用


SELECT
    article_id,
    sumArrayIfOrNull(times, length(sections) > 8)
FROM article_reads
GROUP BY article_id

┌─article_id─┬─sumArrayOrNullIf(times, greater(length(sections), 8))─┐
│         14 │                                                  4779 │
│         18 │                                                  3001 │
│         19 │                                                  NULL │
...
│         17 │                                                 14424 │
└────────────┴───────────────────────────────────────────────────────┘

我们使用了 sumArrayIfOrNull() 函数来计算阅读超过八个部分的文章的总时间。请注意，对于阅读超过八个部分的案例为零的文章，将返回 NULL，因为我们也使用了 OrNull() 组合器。

如果我们结合使用数组函数和组合器，我们可以处理更高级的案例


SELECT
    article_id,
    countArray(arrayFilter(x -> (x > 120), times)) AS sections_engaged
FROM article_reads
GROUP BY article_id

┌─article_id─┬─sections_engaged─┐
│         14 │               26 │
│         18 │               44 │
...
│         17 │               98 │
└────────────┴──────────────────┘

在这里，我们首先使用 arrayFilter 函数过滤 times 数组，以删除所有低于 120 秒的值。然后，我们使用 countArray 来计算每篇文章的过滤时间（在我们的例子中，这意味着参与阅读）。

ClickHouse 中提供的另一种强大类型是 Map。与数组一样，我们可以使用 Map() 组合器将聚合应用于此类型。

假设我们有下表，其中包含 Map 列类型


CREATE TABLE page_loads
(
    `time` DateTime,
    `url` String,
    `params` Map(String, UInt32)
)
ENGINE = MergeTree
ORDER BY (url, time)

┌────────────────time─┬─url─┬─params───────────────────────────────┐
│ 2023-01-25 17:44:26 │ /   │ {'load_speed':100,'scroll_depth':59} │
│ 2023-01-25 17:44:37 │ /   │ {'load_speed':400,'scroll_depth':12} │
└─────────────────────┴─────┴──────────────────────────────────────┘

我们可以对 sum() 和 avg() 函数使用 Map() 组合器，以获取总加载时间和平均滚动深度


SELECT
    sumMap(params)['load_speed'] AS total_load_time,
    avgMap(params)['scroll_depth'] AS average_scroll
FROM page_loads

┌─total_load_time─┬─average_scroll─┐
│             500 │           35.5 │
└─────────────────┴────────────────┘

Map() 组合器也可以与其他组合器一起使用


SELECT sumMapIf(params, url = '/404')['scroll_depth'] AS average_scroll FROM page_loads

处理数组列的另一种方法是聚合来自两个数组的对应值。这将生成另一个数组。这可以用于向量化数据（如向量或矩阵），并通过 ForEach() 组合器实现

ForEach combinator

假设我们有下表，其中包含向量


SELECT * FROM vectors

┌─title──┬─coordinates─┐
│ first  │ [1,2,3]     │
│ second │ [2,2,2]     │
│ third  │ [0,2,1]     │
└────────┴─────────────┘

要计算平均坐标数组（向量），我们可以使用 avgForEach() 组合函数


SELECT avgForEach(coordinates) FROM vectors

┌─avgForEach(coordinates)─┐
│ [1,2,2]                 │
└─────────────────────────┘

这将要求 ClickHouse 计算所有 coordinates 数组的第一个元素的平均值，并将其放入结果数组的第一个元素中。然后对第二个和第三个元素重复相同的操作。

当然，也支持与其他组合器一起使用


SELECT avgForEachIf(coordinates, title != 'second') FROM vectors

┌─avgForEachIf(coordinates, notEquals(title, 'second'))─┐
│ [0.5,2,2]                                             │
└───────────────────────────────────────────────────────┘

ClickHouse 允许使用中间聚合状态而不是结果值。假设我们需要计算我们案例中的唯一值，但我们不想保存值本身（因为它占用空间）。在这种情况下，我们可以对 uniq() 函数使用 State() 组合器以保存中间聚合状态，然后使用 Merge() 组合器来计算实际值


SELECT uniqMerge(u)
FROM
(
    SELECT uniqState(number) AS u FROM numbers(5)
    UNION ALL
    SELECT uniqState(number + 1) AS u FROM numbers(5)
)

┌─uniqMerge(u)─┐
│            6 │
└──────────────┘

在这里，第一个嵌套查询将返回 1...5 数字的唯一计数的状态。第二个嵌套查询返回 2...6 数字的相同状态。父查询然后使用 uniqMerge() 函数来合并我们的状态并获取我们看到的所有唯一数字的计数

uniqState() and uniqMerge() examples

我们为什么要这样做？仅仅是因为聚合状态比原始数据占用更少的空间。当我们想要将此状态存储在磁盘上时，这一点尤其重要。例如，uniqState() 数据比 100 万个整数占用少 15 倍的空间


SELECT
    table,
    formatReadableSize(total_bytes) AS size
FROM system.tables
WHERE table LIKE 'numbers%'

┌─table─────────┬─size───────┐
│ numbers       │ 3.82 MiB   │ <- we saved 1 million ints here
│ numbers_state │ 245.62 KiB │ <- we save uniqState for 1m ints here
└───────────────┴────────────┘

ClickHouse 提供了一个 AggregatingMergeTree 表引擎，用于存储聚合状态并在主键上自动合并它们。让我们创建一个表来存储来自先前示例的每日付款的聚合数据


CREATE TABLE payments_totals
(
    `date` Date,
    `total_amount` AggregateFunction(sum, Float)
)
ENGINE = AggregatingMergeTree
ORDER BY date

我们使用了 AggregateFunction 类型，以使 ClickHouse 知道我们将存储聚合总状态而不是标量值。在插入时，我们需要使用 sumState 函数来插入聚合状态


INSERT INTO payments_totals SELECT
    date(create_time) AS date,
    sumState(total_amount)
FROM payments
WHERE status = 'confirmed'
GROUP BY date

最后，我们需要使用 sumMerge() 函数来获取结果值


┌─sumMerge(total_amount)─┐
│     12033.219916582108 │
└────────────────────────┘

注意，ClickHouse 提供了一种基于物化视图轻松使用聚合表引擎的方法。ClickHouse 还提供了一个 SimpleState 组合器，作为可与某些聚合函数（如“sum”或“min”）一起使用的优化版本。

聚合函数组合器为 ClickHouse 中任何数据结构之上的分析查询带来了几乎无限的可能性。我们可以向聚合添加条件，将函数应用于数组元素或获取中间状态以聚合形式存储数据，但仍可用于查询。

分享这篇文章

订阅我们的新闻通讯

随时了解功能发布、产品路线图、支持和云服务！

正在加载表单...

博客 / 工程

在 ClickHouse 中使用聚合组合器

如何使用组合器

向聚合添加条件

仅聚合唯一条目

组合 `Distinct` 和 `If`

在聚合之前将数据拆分为组

控制空结果的聚合值

与其他组合器一起使用

聚合数组

将 `Array` 与其他组合器一起使用

聚合 Map

聚合相应的数组值

使用聚合状态

总结

订阅我们的新闻通讯

最新文章

博客 / 工程

在 ClickHouse 中使用聚合组合器

如何使用组合器

向聚合添加条件

仅聚合唯一条目

组合 Distinct 和 If

在聚合之前将数据拆分为组

控制空结果的聚合值

与其他组合器一起使用

聚合数组

将 Array 与其他组合器一起使用

聚合 Map

聚合相应的数组值

使用聚合状态

总结

订阅我们的新闻通讯

最新文章

组合 `Distinct` 和 `If`

将 `Array` 与其他组合器一起使用