【大数据系列】Doris模型介绍-聚合模型Aggregate

努力成为一个程序猿.

703人浏览 · 2025-10-14 15:12:14

努力成为一个程序猿. · 2025-10-14 15:12:14 发布

Doris 的聚合模型专为高效处理大规模数据查询中的聚合操作设计，它通过预聚合数据，减少重复计算，提升查询性能。聚合模型只存储聚合后的数据，节省存储空间并加速查询。

使用场景：

明细数据汇总，例如数据报表
不需要查询明细数据，例如只需要保存根据某些维度的最新数据

原理

聚合模型的表中的列按照是否设置了 AggregationType，分为 Key（维度列）和 Value（指标列）。

没有设置 AggregationType 的称为 Key，设置了 AggregationType 的称为 Value。

导入数据时，key列相同的行会聚合成一行。而value列会按照设置的AggregateType进行聚合。

聚合方式	描述
SUM	求和，多行的 Value 进行累加。
REPLACE	替代，下一批数据中的 Value 会替换之前导入过的行中的 Value。
MAX	保留最大值。
MIN	保留最小值。
REPLACE_IF_NOT_NULL	非空值替换。与 REPLACE 的区别在于对 null 值，不做替换。
HLL_UNION	HLL 类型的列的聚合方式，通过 HyperLogLog 算法聚合。
BITMAP_UNION	BITMAP 类型的列的聚合方式，进行位图的并集聚合。

1）每一批次数据导入的 ETL 阶段。该阶段会在每一批次导入的数据内部进行聚合。

（2）底层 BE 进行数据 Compaction 的阶段。该阶段，BE 会对已导入的不同批次的数据进行进一步的聚合。

（3）数据查询阶段。在数据查询时，对于查询涉及到的数据，会进行对应的聚合。

数据在不同时间，可能聚合的程度不一致。比如一批数据刚导入时，可能还未与之前已存在的数据进行聚合。但是对于用户而言，用户只能查询到聚合后的数据。即不同的聚合程度对于用户查询而言是透明的。用户需始终认为数据以最终的完成的聚合程度存在，而不应假设某些聚合还未发生。

建表说明

1)主键必须包含全部的key维度列。否则会报错，

failed to execute sql: agg_keys table should specify aggregate type

2)主键的顺序必须是key维度列的前缀顺序

Key columns should be a ordered prefix of the schema. KeyColumns[1] (starts from zero) is aa, but corresponding column is bb in the previous columns declaration

3)分区键必须是key维度列，否则无法建表

create table aggregate_test
(
pd_id int NOT NULL COMMENT "id",
pd_name varchar(32) NOT NULL COMMENT "名称",
dt date NOT NULL COMMENT "周期",
vx double replace COMMENT "指标",
ct int max COMMENT "维度总和计数"
)    
aggregate key(pd_id,pd_name)
partition by list(dt)
(
    partition dt1 values in ("2025-01-01"),
     partition dt1 values in ("2025-01-02")
)
distributed by hash(pd_id) buckets 5;

避免频繁的执行insert into语句，会因为写入频繁导致写堵塞。

聚合模型保存明细数据

如何利用聚合模型保存明细数据呢？

具体实现方式在选择key列时，选择唯一不重复的键当作key列，这样就可以利用聚合模型Aggregate保存明细数据了，生产不建议这么使用。因为有其他模型可以完成明细数据的保存。

create table aggregate_test
(
pd_id int NOT NULL COMMENT "id",
pd_name varchar(32) NOT NULL COMMENT "名称",
timestamp int
dt date NOT NULL COMMENT "周期",
vx double replace COMMENT "指标",
ct int max COMMENT "维度总和计数"
)    
aggregate key(pd_id,pd_name)
partition by list(dt)
(
    partition dt1 values in ("2025-01-01"),
     partition dt1 values in ("2025-01-02")
)
distributed by hash(pd_id) buckets 5;