第一部分 批处理与流计算概述

1. 批量计算与实时计算的区别

大数据的计算模式按数据的"时间属性"可分为两大类：

对比项 批量计算（批处理） 实时计算（流计算）
数据形态 有界数据集，数据先落盘再计算 无界数据流，数据持续不断到达
处理时机 先存储，后计算（离线） 边到达边计算（在线）
延迟 分钟级、小时级甚至天级 毫秒级、秒级
数据规模 单批数据量大 单条/微批数据量小，但总量持续增长
计算次数 一份数据算一次，可重复算 数据流过一次即计算，通常"一次通过"
典型作业 MapReduce 作业、Spark 批作业、Hive SQL Storm Topology、Spark Streaming、Flink Job
关注指标 吞吐量（Throughput） 延迟（Latency）与实时性
作业生命周期 有开始有结束，跑完即退出 长驻运行，7×24 不停机

要点理解：

  • 批量计算是"数据等程序"——数据攒够一批，作业启动，算完输出结果后退出。
  • 流计算是"程序等数据"——作业先启动并常驻，数据来一条处理一条（或攒成很小的一批处理一次）。
  • 二者不是互相替代的关系，实际生产中常采用批流结合的架构：流计算保证实时性，批处理保证结果的完整与准确。

2. 流计算的应用场景与处理流程

2.1 应用场景

流计算面向持续产生的实时数据流进行低延迟的连续计算。典型场景包括：

  • 实时网站流量统计、PV/UV 实时看板、实时热词/热搜（如 Storm starter 中的 RollingTopWords）。
  • 实时日志分析与监控告警（服务器指标、应用错误日志的秒级告警）。
  • 金融风控与实时反欺诈（交易发生的瞬间判断风险）。
  • 实时推荐与实时用户画像更新。
  • 物联网、传感器数据的实时采集与异常检测。
  • 交通流量、电力负荷等实时监测调度。

2.2 处理流程

流计算系统的处理流程一般分为三个阶段：

  1. 数据采集（数据实时采集）

    • 从数据源持续、实时地采集数据，要求高可靠、低延迟。
    • 数据源包括日志文件、数据库变更、消息队列、传感器、埋点上报等。
    • 常用采集/传输组件：Flume（日志采集）、Kafka（消息中间件/缓冲）、Scribe 等。
    • 采集端通常只负责"拿到并发送"，不做复杂计算，以保证吞吐。

  2. 数据处理（数据实时计算）

    • 由流计算框架对流入的数据做过滤、转换、聚合、关联、窗口统计等操作。
    • 计算是持续的：作业常驻内存，数据到达即触发计算，结果随时更新。
    • 该阶段由 Storm、Spark Streaming、Flink、Samza 等框架承担。

  3. 结果输出（实时查询服务）

    • 计算结果实时推送到下游：写入内存数据库/缓存（Redis）、NoSQL（HBase）、关系数据库，或直接推送到前端看板、告警系统。
    • 与批处理"用户主动查询已算好的结果"不同，流计算是"结果主动推送给用户"，用户看到的是不断更新的最新值。

在本文的 Storm WordCount 实例中，这三步分别对应：Spout 不断产生英文句子（采集）→ Bolt 切词并累加计数（处理）→ Bolt 输出当前单词的累计次数（结果输出）。

3. 常见流计算框架对比

框架 来源/实现语言 处理模型 延迟 语义保证 特点
Storm Twitter 开源，Clojure/Java 逐条处理（Tuple 级），Topology = Spout + Bolt 毫秒级，真正的实时 至少一次（At-least-once）；Trident 可做到恰好一次 纯流式、延迟最低、编程模型简单；早期无内置状态管理与窗口
Spark Streaming Spark 生态，Scala/Java 微批（Micro-batch），把流切成小批 RDD 秒级（受批间隔限制） 恰好一次（Exactly-once） 与 Spark 批处理、SQL、MLlib 共用一套 API 与集群，吞吐高，便于批流统一
Flink Apache，Java/Scala 原生流处理（批是流的特例） 毫秒级 恰好一次 事件时间（Event Time）、Watermark、丰富的窗口与状态管理、Checkpoint 容错，功能最完整
Samza LinkedIn 开源，Java/Scala 逐条处理，基于 Kafka + YARN 秒级 至少一次 与 Kafka 深度绑定，用 Kafka 做消息传输和缓冲，本地状态存储，运维简单

选型建议：

  • 追求极低延迟、逻辑相对简单：Storm。
  • 已有 Spark 技术栈、希望批流复用同一套代码和集群、可接受秒级延迟：Spark Streaming。
  • 需要精确一次语义、复杂窗口、事件时间处理和大状态：Flink。
  • 数据全部走 Kafka、运行在 YARN 上、希望架构简单：Samza。

第二部分 Spark 与图计算

4. Spark 概述与相比 MapReduce 的优势

Spark 是一个通用的大数据快速计算引擎，可以运行在 Standalone、YARN、Mesos 等资源管理框架之上，并能读写 HDFS 等分布式存储。

相比 Hadoop MapReduce，Spark 的主要优势：

  1. 内存计算

    • MapReduce 每一步计算的中间结果都要写回 HDFS，下一步再读回来，磁盘 I/O 开销巨大。
    • Spark 把中间结果尽可能保存在内存中（RDD 可缓存/持久化），迭代计算与交互式查询时性能提升明显。
    • 对机器学习、图计算这类需要反复迭代同一份数据的算法，优势尤为突出。

  2. DAG 执行引擎

    • MapReduce 只有 Map 和 Reduce 两个阶段，复杂逻辑必须拆成多个串行的 MapReduce 作业，每个作业都要重新启动、重新读写磁盘。
    • Spark 把整个作业组织成 DAG（有向无环图），由 DAGScheduler 划分 Stage、TaskScheduler 调度 Task，可对多个操作做流水线（pipeline）优化，减少不必要的落盘和 Shuffle。

  3. 编程模型丰富

    • MapReduce 只提供 map/reduce 两个原语；Spark 提供 map、filter、flatMap、reduceByKey、groupByKey、join、sortBy 等大量转换（Transformation）与行动（Action）算子。
    • 支持 Scala、Java、Python、R 多语言 API，代码量通常远小于等价的 MapReduce 程序。

  4. 容错机制

    • 基于 RDD 的血统（Lineage）重算，无需像 MapReduce 那样完全依赖中间结果落盘来保证容错。

  5. 一套引擎多种负载

    • 批处理、SQL 查询、流计算、机器学习、图计算可以在同一引擎、同一集群、同一份数据上完成，避免为每种计算维护一套系统。

5. Spark 生态组件

Spark 以 Spark Core 为核心，向上构建了一整套组件栈：

组件 作用
Spark Core 核心引擎，提供 RDD 抽象、任务调度（DAGScheduler/TaskScheduler）、内存管理、故障恢复、与存储系统交互等基础能力，其他组件都建立在它之上
Spark SQL 结构化数据处理，提供 DataFrame/Dataset API 与 SQL 查询能力，可读取 Hive、Parquet、JSON、JDBC 等数据源，内置 Catalyst 优化器
Spark Streaming 流计算组件，采用微批（DStream）模型把实时数据流切成一系列小批 RDD 处理，可与批处理逻辑复用代码
MLlib 分布式机器学习库，提供分类、回归、聚类、协同过滤、降维等算法以及特征处理、模型评估、Pipeline 等工具
GraphX 图计算组件，在 Spark 上提供属性图抽象、常用图算法（PageRank、连通分量、三角计数等）以及 Pregel 风格的迭代计算 API

注意：本文实操部分所依据的源讲义中未包含 GraphX/GraphFrames 的具体安装与代码示例，图计算实操以 Hama（BSP 模型） 为例展开。

6. 图计算模型概述

6.1 Pregel 模型

Pregel 是 Google 提出的大规模图计算模型，其核心思想是 以顶点为中心（vertex-centric）的迭代计算：

  • 图被划分到多台机器上，每个顶点维护自己的状态（value）和出边。
  • 计算按 超步（superstep） 进行：在每个超步中，每个活跃顶点接收上一超步发来的消息、更新自身状态、并向邻居顶点发送消息；所有顶点完成后进行一次全局同步，进入下一超步。
  • 顶点可以主动"投票停止"（vote to halt）进入非活跃状态；当所有顶点都非活跃且没有消息在传递时，整个计算结束。
  • 容错通过在超步边界做 Checkpoint 实现。

这种"计算—通信—同步"的循环结构，正是 BSP（Bulk Synchronous Parallel，整体同步并行）模型。PageRank 是 Pregel 模型最经典的应用：每个顶点在每个超步把自己的 rank 值按出边平分发给邻居，再根据收到的值更新自身 rank，迭代至收敛。

6.2 Hama 图计算模型

Hama 是 Google Pregel 的开源实现。与 Hadoop 适合分布式大数据处理不同，Hama 主要用于分布式的矩阵、graph（图）、网络算法的计算。

简单说，Hama 是在 HDFS 上实现的 BSP（Bulk Synchronous Parallel，整体同步并行） 计算框架，用于弥补 Hadoop 在计算能力上的不足。

本部分主要介绍 Hama 的单机模式安装配置，以及用 Hama 解决一些算法问题。

Hama 中的关键角色与概念：

  • BSPMaster：负责作业调度与超步同步控制，地址由 bsp.master.address 指定。
  • GroomServer：实际执行 BSP Task 的工作进程，分布式环境下在 conf/groomservers 中配置。
  • ZooKeeper：负责各 BSP Peer 之间的同步（barrier synchronization）。
  • BSPPeer：任务执行时的编程接口，提供 send()（发消息）、sync()（同步屏障）、getCurrentMessage()（取消息）、write()（输出）等方法。
  • superstep（超步）：一轮"本地计算 + 消息通信 + 全局同步"。

7. 环境准备与前置条件

  1. 安装好合适版本的 JDK 和 Hadoop。
  2. 获得安装包 hama-dist-0.7.1.tar.gz。

8. 解压安装与配置（单机模式）

8.1 解压

解压到路径 /usr/local/hama：

tar -zxvf hama-dist-0.7.1.tar.gz -C /usr/local

8.2 配置 hama-env.sh

编辑 /usr/local/hama/conf/hama-env.sh，设置 JAVA_HOME：

vi /usr/local/hama/conf/hama-env.sh
export JAVA_HOME=/usr/local/jdk

8.3 配置 hama-site.xml

编辑 /usr/local/hama/conf/hama-site.xml：

vim /usr/local/hama/conf/hama-site.xml
<configuration>
    <property>
      <name>bsp.master.address</name>
      <value>local</value>
      <description>The address of the bsp master server. Either the
      literal string "local" or a host:port for distributed mode
      </description>
    </property>

    <property>
      <name>fs.default.name</name>
      <value>file:///</value>
      <description>
        The name of the default file system. Either the literal string
        "local" or a host:port for HDFS.
      </description>
    </property>

    <property>
      <name>hama.zookeeper.quorum</name>
      <value>localhost</value>
      <description>Comma separated list of servers in the ZooKeeper Quorum.
      For example, "host1.mydomain.com,host2.mydomain.com,host3.mydomain.com".
      By default this is set to localhost for local and pseudo-distributed modes
      of operation. For a fully-distributed setup, this should be set to a full
      list of ZooKeeper quorum servers. If HAMA_MANAGES_ZK is set in hama-env.sh
      this is the list of servers which we will start/stop zookeeper on.
      </description>
    </property>
  </configuration>

配置项说明：

配置项 说明
bsp.master.address BSP 中 BSPMaster 的地址和端口。单机模式设为 local（本机地址）
fs.default.name Hadoop 中 NameNode 的地址和端口，Hama 要用到 Hadoop 的 HDFS。需特别注意
hama.zookeeper.quorum ZooKeeper 相关配置，单机模式下设为本机地址，端口一般固定

另外，conf 文件夹下还有一个 groomservers 文件，用于在分布式环境下配置 groomserver 的地址；单机模式下无需配置，其默认值为 localhost。也可以在 ~/.bashrc 中添加 Hama 的环境变量，这样每次启动就不用转到相应目录下了。

9. 运行示例程序与验证：PageRank

9.1 第一步：生成 randomgraph

运行如下命令：

cd /usr/local/hama/bin/hama
/usr/local/hama/bin/hama jar hama-examples-0.7.1.jar gen fastgen -v 100 -e 10 -o randomgraph -t 2

生成的文件位于 /usr/local/hama 下的 randomgraph，表示 100 个节点、1000 条边的数据，存储在两个文件中：

cat /usr/local/hama/randomgraph/part-00000
cat /usr/local/hama/randomgraph/part-00001

9.2 第二步：执行 PageRank

/usr/local/hama/bin/hama jar hama-examples-0.7.1.jar pagerank -i randomgraph -o pagerankresult -t 4

运行输出（节选）：

19/01/06 21:09:24 INFO bsp.BSPJobClient: Current supersteps number: 61
19/01/06 21:09:24 INFO bsp.BSPJobClient: The total number of supersteps: 61
19/01/06 21:09:24 INFO bsp.BSPJobClient: Counters: 11
19/01/06 21:09:24 INFO bsp.BSPJobClient:   org.apache.hama.bsp.BSPPeerImpl$PeerCounter
19/01/06 21:09:24 INFO bsp.BSPJobClient:     TIME_IN_SYNC_MS=633
19/01/06 21:09:24 INFO bsp.BSPJobClient:     TASK_OUTPUT_RECORDS=100
19/01/06 21:09:24 INFO bsp.BSPJobClient:     SUPERSTEP_SUM=248
19/01/06 21:09:24 INFO bsp.BSPJobClient:     TOTAL_MESSAGES_SENT=12056
19/01/06 21:09:24 INFO bsp.BSPJobClient:     IO_BYTES_READ=3191
19/01/06 21:09:24 INFO bsp.BSPJobClient:     TOTAL_MESSAGES_RECEIVED=11672
19/01/06 21:09:24 INFO bsp.BSPJobClient:     TASK_INPUT_RECORDS=100
19/01/06 21:09:24 INFO bsp.BSPJobClient:   org.apache.hama.bsp.JobInProgress$JobCounter
19/01/06 21:09:24 INFO bsp.BSPJobClient:     LAUNCHED_TASKS=4
19/01/06 21:09:24 INFO bsp.BSPJobClient:     SUPERSTEPS=61
19/01/06 21:09:24 INFO bsp.BSPJobClient:   org.apache.hama.graph.GraphJobRunner$GraphJobCounter
19/01/06 21:09:24 INFO bsp.BSPJobClient:     INPUT_VERTICES=100
19/01/06 21:09:24 INFO bsp.BSPJobClient:     ITERATIONS=31
Job Finished in 12.364 seconds

输出中的计数器正好印证了 BSP/Pregel 模型：SUPERSTEPS=61 表示共执行了 61 个超步，ITERATIONS=31 表示 PageRank 迭代了 31 轮，TOTAL_MESSAGES_SENT / TOTAL_MESSAGES_RECEIVED 是顶点之间传递的消息总数，TIME_IN_SYNC_MS 是花在同步屏障上的时间。

单机模式下，数据读取都在本地文件系统，不需要读取 HDFS 中的文件。

10. 图计算示例代码与运行：计算 PI

10.1 第一步：设置 CLASSPATH 路径

export CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath):/usr/local/jdk/lib/*:/usr/local/hama/*:/usr/local/hama/lib/*

10.2 第二步：编译打包

javac -classpath $CLASSPATH PiEstimator.java

10.3 第三步：执行程序

/usr/local/hama/bin/hama jar PiEstimator.jar PiEstimator

10.4 Java 代码（PiEstimator.java）

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.ClusterStatus;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.NullInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;

public class PiEstimator
{
    private static Path TMP_OUTPUT = new Path("/tmp/pi-"
        + System.currentTimeMillis());

    public static class MyEstimator
        extends
        BSP<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable>
    {
    public static final Log LOG = LogFactory.getLog(MyEstimator.class);
    private String masterTask;
    private static final int iterations = 10000;

    @Override
    public void bsp(
        BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer)
        throws IOException, SyncException, InterruptedException
    {

        int in = 0;
        for (int i = 0; i < iterations; i++)
        {
        double x = 2.0 * Math.random() - 1.0, y = 2.0 * Math.random() - 1.0;
        if ((Math.sqrt(x * x + y * y) < 1.0))
        {
            in++;
        }
        }

        double data = 4.0 * in / iterations;

        peer.send(masterTask, new DoubleWritable(data));
        peer.sync();
    }

    @Override
    public void setup(
        BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer)
        throws IOException
    {
        // Choose one as a master
        this.masterTask = peer.getPeerName(peer.getNumPeers() / 2);
    }

    @Override
    public void cleanup(
        BSPPeer<NullWritable, NullWritable, Text, DoubleWritable, DoubleWritable> peer)
        throws IOException
    {
        if (peer.getPeerName().equals(masterTask))
        {
        double pi = 0.0;
        int numPeers = peer.getNumCurrentMessages();
        DoubleWritable received;
        while ((received = peer.getCurrentMessage()) != null)
        {
            pi += received.get();
        }

        pi = pi / numPeers;
        peer.write(new Text("Estimated value of PI is"),
            new DoubleWritable(pi));
        }
    }
    }

    static void printOutput(HamaConfiguration conf) throws IOException
    {
    FileSystem fs = FileSystem.get(conf);
    FileStatus[] files = fs.listStatus(TMP_OUTPUT);
    for (int i = 0; i < files.length; i++)
    {
        if (files[i].getLen() > 0)
        {
        FSDataInputStream in = fs.open(files[i].getPath());
        IOUtils.copyBytes(in, System.out, conf, false);
        in.close();
        break;
        }
    }

    fs.delete(TMP_OUTPUT, true);
    }

    public static void main(String[] args) throws InterruptedException,
        IOException, ClassNotFoundException
    {
    // BSP job configuration
    HamaConfiguration conf = new HamaConfiguration();

    BSPJob bsp = new BSPJob(conf, PiEstimator.class);
    // Set the job name
    bsp.setJobName("Pi Estimation Example");
    bsp.setBspClass(MyEstimator.class);
    bsp.setInputFormat(NullInputFormat.class);
    bsp.setOutputKeyClass(Text.class);
    bsp.setOutputValueClass(DoubleWritable.class);
    bsp.setOutputFormat(TextOutputFormat.class);
    FileOutputFormat.setOutputPath(bsp, TMP_OUTPUT);

    BSPJobClient jobClient = new BSPJobClient(conf);
    ClusterStatus cluster = jobClient.getClusterStatus(true);

    if (args.length > 0)
    {
        bsp.setNumBspTask(Integer.parseInt(args[0]));
    } else
    {
        // Set to maximum
        bsp.setNumBspTask(cluster.getMaxTasks());
    }

    long startTime = System.currentTimeMillis();
    if (bsp.waitForCompletion(true))
    {
        printOutput(conf);
        System.out.println("Job Finished in "
            + (System.currentTimeMillis() - startTime) / 1000.0
            + " seconds");
    }
    }
}

10.5 代码要点说明

  • MyEstimator 继承自 BSP，核心方法为 bsp()：每个 task 在 [-1,1] 正方形内随机生成 iterations（10000）个点，统计落在单位圆内的点数 in，用 4.0 * in / iterations 估算 π，然后将结果发送给 master task 并调用 peer.sync() 进入下一个超步（superstep）同步。
  • setup()：在所有 peer 中选取一个作为 master（取中间编号的 peer）。
  • cleanup()：master 收集所有 task 发来的估算值，求平均后输出最终的 π 估计值。
  • main()：配置 BSP Job（输入格式为 NullInputFormat、输出为 TextOutputFormat），可通过命令行参数指定 task 数，未指定时使用集群最大 task 数。

练习

  1. 用表格形式对比批量计算与实时计算，至少列出 5 项差异（延迟、数据形态、作业生命周期、关注指标、典型框架）。

  2. 列举 Spark 相比 MapReduce 的至少三点优势，并解释"内存计算"与"DAG 执行引擎"为什么能提升迭代计算的性能。

  3. 说明 Spark 生态中 Spark Core、Spark SQL、Spark Streaming、MLlib、GraphX 各自的职责。

  4. 简述 Pregel 模型的"以顶点为中心 + 超步同步"的计算过程，并说明 BSP 模型中"同步屏障"的作用。

  5. 在单机环境完成 Hama 的安装配置：解压 hama-dist-0.7.1.tar.gz 到 /usr/local/hama，配置 hama-env.sh 的 JAVA_HOME 与 hama-site.xml 的三个属性，并说明每个属性的含义。

  6. 运行 Hama 的 PageRank 实例：先用 gen fastgen 生成 100 个节点、1000 条边的随机图，再执行 pagerank，记录输出中的 SUPERSTEPS、ITERATIONS、TOTAL_MESSAGES_SENT 并解释其含义。

  7. 编译并运行 PiEstimator.java，说明 bsp()、setup()、cleanup() 三个方法各自完成了什么工作，以及 peer.send() 与 peer.sync() 的配合关系。

  8. 修改 PiEstimator.java 中的 iterations 值（如改为 100000），重新编译运行，观察 π 估算精度的变化。

最后修改：2026 年 09 月 20 日
© 允许规范转载
如果觉得我的文章对你有用，请随意赞赏