数据库重复聚合问题咨询：如何避免数据重复统计与丢失

碰到这种重复聚合的问题太常见了，我帮你梳理几个既能避免重复又不会丢数据的方案，顺便聊聊你提到的「记录上次聚合时间」方式的坑——大概率是怕延迟写入的数据被漏掉对吧？毕竟很多场景下，数据不是实时写入的，比如有些日志可能过了半小时才同步到数据库，这时候只取上次聚合时间之后的数据，就会漏掉那些时间戳早但刚进来的数据，导致统计结果不全。

下面是几个实用的解决方案，你可以根据业务场景选择：

方案一：用UPSERT替换单纯INSERT（最省心） #

核心思路是：不管跑多少次查询，只要ts（分钟级时间戳）已经存在于聚合表，就更新统计值，而不是插入新记录。这需要给aggregation_table的ts字段加主键或唯一约束。

以PostgreSQL为例，写法如下：

INSERT INTO aggregation_table (ts, count_value)
SELECT 
    CAST(EXTRACT('epoch' FROM timestamp)/60 AS BIGINT)*60 AS ts, 
    COUNT(aggregated_value) AS count_value
FROM aggregated_table
GROUP BY ts
ON CONFLICT (ts) DO UPDATE 
SET count_value = aggregation_table.count_value + EXCLUDED.count_value;

如果是MySQL，用ON DUPLICATE KEY UPDATE：

INSERT INTO aggregation_table (ts, count_value)
SELECT 
    UNIX_TIMESTAMP(timestamp) DIV 60 * 60 AS ts, 
    COUNT(aggregated_value) AS count_value
FROM aggregated_table
GROUP BY ts
ON DUPLICATE KEY UPDATE 
count_value = aggregation_table.count_value + VALUES(count_value);

优缺点 #

• ✅ 优点：实现简单，不需要额外维护任何状态；哪怕有延迟写入的数据，再次执行查询时会自动累加对应ts的统计值，不会丢数据，也不会产生重复记录。
• ❌ 缺点：如果aggregated_table数据量很大，每次全表扫描聚合可能性能不高，适合数据量中等或者能接受全表扫描的场景。

方案二：标记已处理记录（性能最优） #

如果aggregated_table数据量很大，全表扫描太耗时，可以给原表加一个标记字段，或者新增一张表记录已处理的主键，确保每次只处理未统计过的数据。

这里以给原表加is_processed字段为例：

1. 先给原表添加标记字段：

ALTER TABLE aggregated_table ADD COLUMN is_processed BOOLEAN DEFAULT FALSE;

2. 每次聚合时只处理未标记的记录，处理完后标记为已处理：

WITH to_process AS (
    SELECT id, timestamp, aggregated_value
    FROM aggregated_table
    WHERE is_processed = FALSE
    FOR UPDATE -- 加锁防止并发处理时重复统计
)
INSERT INTO aggregation_table (ts, count_value)
SELECT 
    CAST(EXTRACT('epoch' FROM timestamp)/60 AS BIGINT)*60 AS ts, 
    COUNT(aggregated_value) AS count_value
FROM to_process
GROUP BY ts
ON CONFLICT (ts) DO UPDATE 
SET count_value = aggregation_table.count_value + EXCLUDED.count_value;

-- 标记已处理的记录
UPDATE aggregated_table
SET is_processed = TRUE
WHERE id IN (SELECT id FROM to_process);

优缺点 #

• ✅ 优点：每次只处理新数据，聚合速度快；完全不会漏统计延迟数据（哪怕数据时间戳早，只要没标记就会被处理）。
• ❌ 缺点：需要修改原表结构（或者新增关联表），增加了维护成本；如果原表有删除操作，需要同步处理标记状态。

方案三：时间窗口+缓冲区（不修改原表的折中方案） #

如果不能修改原表，又想兼顾性能和数据完整性，可以用「上次聚合时间+缓冲区」的方式，每次聚合时不仅处理上次聚合时间之后的数据，还会重新处理缓冲区时间范围内的数据（比如1小时），再用UPSERT更新统计值，这样就能覆盖延迟写入的数据。

步骤如下：

1. 先创建一张元数据表记录聚合状态：

CREATE TABLE IF NOT EXISTS aggregation_metadata (
    id INT PRIMARY KEY DEFAULT 1,
    last_processed_ts BIGINT NOT NULL DEFAULT 0 -- 存储上次聚合到的时间戳（秒级）
);
-- 初始化数据（如果表为空）
INSERT INTO aggregation_metadata (last_processed_ts) VALUES (0) ON CONFLICT (id) DO NOTHING;

2. 执行聚合逻辑：

WITH current_metadata AS (
    SELECT last_processed_ts FROM aggregation_metadata WHERE id = 1
),
to_aggregate AS (
    SELECT 
        CAST(EXTRACT('epoch' FROM timestamp)/60 AS BIGINT)*60 AS ts, 
        COUNT(aggregated_value) AS count_value
    FROM aggregated_table
    -- 处理上次聚合时间前1小时到当前时间的数据，覆盖延迟写入
    WHERE EXTRACT('epoch' FROM timestamp) > (SELECT last_processed_ts FROM current_metadata) - 3600
    GROUP BY ts
),
update_aggregation AS (
    INSERT INTO aggregation_table (ts, count_value)
    SELECT ts, count_value FROM to_aggregate
    ON CONFLICT (ts) DO UPDATE 
    SET count_value = aggregation_table.count_value + EXCLUDED.count_value
)
-- 更新上次聚合时间为当前时间
UPDATE aggregation_metadata
SET last_processed_ts = EXTRACT('epoch' FROM NOW())::BIGINT
WHERE id = 1;

你可以根据业务调整缓冲区的时间（比如30分钟、2小时），确保延迟数据都能被覆盖。

优缺点 #

• ✅ 优点：不需要修改原表；兼顾性能（大部分时候只处理新数据+小范围缓冲区）和数据完整性（覆盖延迟数据）。
• ❌ 缺点：需要维护元数据表；每次聚合会重新处理缓冲区的数据，但因为用了UPSERT，只会累加正确数值，不会产生重复记录。

内容的提问来源于stack exchange，提问作者Shuumi