第3章 全量计算模型：批处理模式与核心引擎

全量计算，说白了就是「今天不管昨天算过什么，我把所有数据重新算一遍」。听起来有点笨？但很多场景下，这是最稳妥、最可靠的方式。我做了这么多年数据架构，见过不少团队试图用增量省时间，结果数据对不上，最后还是老老实实跑全量。

这一章我们聊聊全量计算的几个核心模式：批处理的思想源头 MapReduce、Spark Core 怎么处理全量数据、以及 Hive 做全量 ETL 的那些坑和技巧。

批处理模式：先有鸡还是先有蛋？

批处理的核心思想很简单：攒一批，算一次。数据不是一条一条来的，而是按时间窗口（比如每天凌晨）把过去一天的数据打包，然后一次性喂给计算引擎。

为什么会这样？因为早期硬件资源有限，IO 是最大的瓶颈。你想想看，如果每条数据都触发一次计算，磁盘寻道时间就够你受的。批处理正好解决了这个问题——数据顺序读取，顺序写入，充分利用磁盘带宽。

批处理的三个基本特征：

  • 数据有界——处理的是一个确定的数据集，不是无限流
  • 延迟较高——分钟级到小时级，不适合实时场景
  • 结果精确——全量重算，不存在「丢数据」的问题

我个人习惯把批处理比作「期末考试」——把所有知识点攒到一块儿考，虽然压力大，但成绩最真实。增量计算更像是「随堂测验」，快是快，但容易漏掉重点。

MapReduce 思想：分而治之的祖师爷

MapReduce 是 Google 在 2004 年提出的编程模型。说实话，现在直接写 MapReduce 的人不多了，但它的思想渗透到了几乎所有大数据框架里。

核心就两个阶段：

  • Map（映射）：把输入数据拆成键值对，每个分片独立处理
  • Reduce（归约）：把相同 key 的数据聚合起来，做最终计算

中间还有个 Shuffle 阶段——这是 MapReduce 最耗时的部分。我在项目中遇到过一个问题：Map 阶段跑得飞快，Reduce 阶段卡死。查了半天，发现是某个 key 的数据量特别大（数据倾斜），所有数据都挤到一个 Reduce 任务里了。

避坑指南：我曾经因为没做数据预处理，导致一个 Reduce 任务跑了 6 小时。后来加了个「二次散列」——在 Map 阶段把热点 key 打散，Reduce 阶段再合并。时间直接降到 40 分钟。

来看一个简单的 WordCount 伪代码，感受一下 MapReduce 的思维：

// Map 阶段：把每行文本拆成单词
map(String key, String value):
    // key: 行号, value: 一行文本
    for each word w in value:
        EmitIntermediate(w, "1");

// Reduce 阶段：把相同单词的计数加起来
reduce(String key, Iterator values):
    // key: 单词, values: 该单词的所有计数
    int result = 0;
    for each v in values:
        result += ParseInt(v);
    Emit(AsString(result));

你看，逻辑就这么简单。但真正在生产环境跑的时候，要考虑的事情多得多——数据分区、Combiner 优化、压缩策略……嗯，这些我们后面会细讲。

Spark Core 全量处理：内存计算的威力

MapReduce 有个硬伤——每个 Job 都要落盘。Map 写完磁盘，Reduce 读磁盘，中间 Shuffle 还要写磁盘。磁盘 IO 成了瓶颈。

Spark 的思路不一样：能放内存就放内存，实在放不下才落盘。这个设计让 Spark 在全量计算场景下比 MapReduce 快 10-100 倍。

Spark Core 的核心抽象是 RDD（弹性分布式数据集）。说白了，RDD 就是一个不可变的、分区的数据集合。你对 RDD 做的每个操作（map、filter、reduceByKey），Spark 都会记录成一个 Lineage（血统），如果某个分区丢了，根据血统重新算就行。

注意：RDD 的惰性求值是个双刃剑。你写了一堆 transformation，Spark 不会立刻执行，只有遇到 action（比如 count、collect）才会触发计算。我见过新手写了个大循环，里面全是 transformation，结果跑了一天没出结果——因为根本没触发计算。

全量处理在 Spark 里最常见的模式是这样的：

// 读取全量数据
val allData = spark.read.parquet("/data/orders/")

// 数据清洗
val cleaned = allData.filter(col("status") =!= "invalid")
                     .dropDuplicates("order_id")

// 业务聚合
val result = cleaned.groupBy("date", "region")
                    .agg(
                        sum("amount").as("total_amount"),
                        count("order_id").as("order_count")
                    )

// 触发计算并写出
result.write.mode("overwrite").parquet("/data/daily_report/")

这段代码看起来简单，但有几个关键点：

  • mode("overwrite")——全量计算一定要覆盖写，不能 append，否则数据会重复
  • dropDuplicates——全量数据里可能有重复，去重是标准操作
  • parquet 格式——列式存储，压缩率高，适合全量扫描

我个人建议，在全量计算场景下，优先用 Spark SQL 而不是 RDD API。为什么？Spark SQL 有 Catalyst 优化器，能自动做谓词下推、列剪枝，你少操很多心。

Hive 全量 ETL：老当益壮的数据仓库

Hive 本质上就是把 SQL 翻译成 MapReduce（或者 Tez、Spark）任务。虽然它不像 Spark 那么「潮」，但在数据仓库领域，Hive 的地位依然稳固。

全量 ETL 在 Hive 里最常见的做法是：

  1. 创建外部表——指向原始数据目录
  2. 写 INSERT OVERWRITE——覆盖目标分区
  3. 分区裁剪——只扫描需要的分区

来看一个典型的全量 ETL 脚本：

-- 创建源表（外部表，指向 HDFS 原始数据）
CREATE EXTERNAL TABLE IF NOT EXISTS ods_orders (
    order_id STRING,
    user_id STRING,
    amount DOUBLE,
    status STRING,
    create_time TIMESTAMP
)
PARTITIONED BY (dt STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
LOCATION '/data/ods/orders/';

-- 全量计算：清洗并写入 dwd 层
INSERT OVERWRITE TABLE dwd_orders PARTITION (dt = '2024-01-15')
SELECT
    order_id,
    user_id,
    amount,
    status,
    create_time
FROM ods_orders
WHERE dt = '2024-01-15'
  AND status IS NOT NULL
  AND amount > 0;

这里有个细节：INSERT OVERWRITE 是全量覆盖，不是追加。如果你用 INSERT INTO，数据会重复。我见过有人把这两个搞混，结果报表数据翻了一倍，排查了一整天。

经验之谈：Hive 全量 ETL 的性能瓶颈通常不在计算，而在小文件。如果源表有大量小文件（比如几百个 1MB 的文件），Hive 启动的 Map 任务数会暴增。我建议在 ETL 之前先做一次小文件合并，或者用 Spark 做预处理。

全量计算的核心流程

下面这张图展示了全量计算从数据源到结果输出的完整链路：

全量计算核心流程 数据源 HDFS / 对象存储 数据读取 全量扫描 / 分区裁剪 计算引擎 MapReduce / Spark 结果输出 覆盖写入 中间处理：数据清洗 → 去重 → 聚合 → 排序 MapReduce 磁盘落盘 适合超大规模 延迟高 稳定性好 Spark Core 内存计算 速度快 10-100x 内存管理复杂 血统容错 Hive SQL 化操作 数据仓库标准 分区管理方便 小文件问题

三种全量计算引擎的对比

我把 MapReduce、Spark Core、Hive 放在一起做了个对比，方便你选型：

维度 MapReduce Spark Core Hive
计算模型 Map + Shuffle + Reduce DAG + RDD Lineage SQL → MapReduce/Tez/Spark
存储介质 磁盘（每次落盘） 内存优先，磁盘兜底 磁盘（HDFS）
适用场景 超大规模、稳定性优先 迭代计算、交互式查询 数据仓库 ETL、报表
开发效率 低（Java 代码量大） 中（Scala/Python API） 高（纯 SQL）
容错机制 Task 级别重试 RDD 血统 + Checkpoint Task 级别重试
数据倾斜处理 需要手动 Combiner 自适应查询执行（AQE） MapJoin / SkewJoin 优化

选型建议：别盲目追新。如果你的数据量在 TB 级以下，Hive + 分区裁剪完全够用。到了 PB 级，Spark 的优势才真正体现出来。MapReduce 现在用得少了，但理解它的思想对排查问题很有帮助——毕竟 Spark 的 Shuffle 原理和 MapReduce 一脉相承。

全量计算的最佳实践

做了这么多年全量计算，我总结了几条铁律：

  • 永远用覆盖写，不用追加写——全量计算的结果是「快照」，不是「流」
  • 分区是性能的命根子——按日期分区，查询时尽量走分区裁剪
  • 数据倾斜要提前预防——热点 key 打散、Salting 技术、两阶段聚合
  • 小文件合并不能省——HDFS 不适合存大量小文件，ETL 后做一次合并
  • 监控中间结果——每个阶段都输出行数，方便定位问题
← 上一章 📖 返回目录 下一章 →