📚 课程目录第4章预聚合的实现方式

4、预聚合的实现方式：基于物化视图、预计算任务与流式计算

聊到预聚合的具体落地，很多同学第一反应就是「不就是提前算好存起来吗？」。嗯，话是没错，但怎么算、在哪算、什么时候算，这里面的门道可不少。我这些年踩过的坑，十有八九都是因为选错了实现方式。

说白了，预聚合的实现就三条路：物化视图、离线批处理、流式计算。它们各有各的脾气，选对了事半功倍，选错了……嗯，咱们一个个来看。

4.1 基于物化视图（Materialized View）

物化视图这东西，我刚开始接触时觉得它就是「带存储的视图」。后来才发现，它其实是数据库帮你自动维护的预聚合结果集。

你想想看，普通视图就是个查询模板，每次访问都要重新算。物化视图不一样——它把结果实实在在地存成一张表，而且数据库会自动帮你增量刷新。

核心思路：在数据库层面声明「我要按什么维度聚合」，数据库负责维护和更新。

4.1.1 适用场景

OLAP 场景：ClickHouse、Doris、Snowflake 等分析型数据库原生支持
维度相对稳定：聚合维度不会频繁变更
查询模式固定：比如按天、按地区统计销售额

4.1.2 实战示例（ClickHouse）

-- 创建物化视图，按天和城市预聚合订单金额
CREATE MATERIALIZED VIEW order_daily_mv
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(day)
ORDER BY (day, city_id)
AS SELECT
    toDate(order_time) AS day,
    city_id,
    sum(amount) AS total_amount,
    count() AS order_count
FROM orders
GROUP BY day, city_id;

这段 SQL 一跑，之后每次往 orders 表写数据，ClickHouse 都会自动把增量数据聚合到 order_daily_mv 里。查询时直接查物化视图，速度能快几十倍。

我的经验：物化视图最适合「写少读多」的场景。我曾经在一个实时看板项目里用了 ClickHouse 的物化视图，把原本 3 秒的查询压到了 50 毫秒以内。但要注意——物化视图不是银弹，维度爆炸时存储会涨得很快。

4.1.3 避坑指南

我曾经在一个项目里对 20 个维度做了全组合物化视图，结果存储膨胀了 30 倍，写入性能直接崩了。后来只保留了 Top-5 高频维度组合，才把问题解决。

不要对所有维度组合都建物化视图
监控刷新延迟，特别是数据量大的时候
物化视图的刷新策略要选对：增量 vs 全量

4.2 基于预计算任务（离线批处理）

如果说物化视图是「数据库帮你干活」，那离线批处理就是「你自己撸起袖子干」。这种方式最灵活，但也最考验架构设计。

我个人的习惯是：能用物化视图解决的，绝不自己写批处理。但现实往往没那么理想——当你的数据源不在同一个数据库，或者聚合逻辑特别复杂时，批处理就成了唯一选择。

4.2.1 典型架构

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  原始数据层   │ ──▶ │  预计算任务    │ ──▶ │  聚合结果层   │
│  (Hive/HDFS) │     │  (Spark/MapReduce) │     │  (MySQL/ES)  │
└─────────────┘     └──────────────┘     └─────────────┘

4.2.2 实现步骤

数据抽取：从业务库或消息队列拉取原始数据
维度对齐：统一时间粒度、维度编码
聚合计算：按预定义的维度组合做 GROUP BY
结果存储：写入高性能查询引擎（如 ES、Druid）

4.2.3 代码示例（Spark SQL）

// 每日凌晨执行预计算任务
val dailyAggDF = spark.sql("""
  SELECT
    date_format(order_time, 'yyyy-MM-dd') AS day,
    city_id,
    product_category,
    sum(amount) AS revenue,
    count(DISTINCT user_id) AS uv
  FROM orders
  WHERE order_time >= '${yesterday}'
    AND order_time < '${today}'
  GROUP BY day, city_id, product_category
""")

dailyAggDF.write
  .mode("overwrite")
  .format("parquet")
  .save("/data/agg/daily/")

关键点：离线批处理的核心是「增量计算」。全量重跑只适合数据量小的场景，否则 T+1 都跑不完。

4.2.4 优缺点对比

维度	物化视图	离线批处理
实时性	秒级~分钟级	T+1 或小时级
灵活性	受限于数据库能力	完全可控
维护成本	低（数据库自动维护）	高（需要调度、监控）
适用数据源	单数据库内	多数据源、异构系统

4.3 基于流式计算（Flink / Spark Streaming）

好，前两种方式都有个共同问题——延迟。物化视图再快也是秒级，批处理更是 T+1。那如果业务要求「数据产生后几秒内就要看到聚合结果」呢？

这时候就得请出流式计算了。我最早用 Flink 做预聚合是在一个实时风控项目里，要求 5 秒内统计出每个用户的交易频次。说实话，第一次跑通时还挺激动的。

4.3.1 流式预聚合的核心逻辑

说白了就是：数据在流动的过程中就完成聚合，而不是等落盘后再算。

// Flink SQL 实现实时预聚合
CREATE TABLE orders_stream (
  order_id STRING,
  user_id STRING,
  amount DOUBLE,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'format' = 'json'
);

-- 每分钟滚动窗口聚合
CREATE TABLE minute_agg_sink (
  window_start TIMESTAMP(3),
  user_id STRING,
  total_amount DOUBLE,
  order_count BIGINT
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://...'
);

INSERT INTO minute_agg_sink
SELECT
  TUMBLE_START(order_time, INTERVAL '1' MINUTE),
  user_id,
  SUM(amount),
  COUNT(*)
FROM orders_stream
GROUP BY
  TUMBLE(order_time, INTERVAL '1' MINUTE),
  user_id;

我个人的习惯：流式预聚合一定要做「两阶段聚合」。先在本机做一次局部聚合，再发送到全局做最终聚合。这样可以大幅减少网络 shuffle 的数据量。

4.3.2 三种方式的对比总结

实现方式	延迟	吞吐	开发成本	典型场景
物化视图	秒~分钟	高	低	OLAP 看板、报表
离线批处理	小时~天	极高	中	T+1 报表、数据仓库
流式计算	秒级	中高	高	实时监控、风控、推荐

4.3.3 选型建议

延迟容忍度 > 1 分钟：优先考虑物化视图，省心省力
需要跨数据源聚合：离线批处理更靠谱
延迟要求 < 10 秒：别犹豫，上 Flink 或 Spark Streaming
团队能力有限：物化视图 > 批处理 > 流式（按维护成本排序）

我曾经在一个项目里强行用 Flink 做 T+1 的离线聚合，结果搞复杂了——既要处理乱序数据，又要管理状态后端，最后还不如直接用 Spark 批处理来得稳。记住：不要为了用新技术而用新技术。

嗯，三种方式各有千秋。我个人的经验是：先想清楚你的延迟要求，再考虑团队能力，最后才是技术选型。很多时候，最简单的方案就是最好的方案。