流计算实战（ZooKeeper + Storm）

11. 实战概述

流计算面向持续产生的实时数据流进行低延迟的连续计算。本部分实践以 Apache Storm 为流计算框架，并以 ZooKeeper 作为其协调服务。运行 Storm 之前需要先启动 ZooKeeper，再依次启动 Storm 的 nimbus 和 supervisor 进程。

启动顺序：ZooKeeper → nimbus → supervisor → 提交 Topology。

12. 安装部署 ZooKeeper

详见 16-06-ZooKeeper集群.md

将 zookeeper-3.4.12.tar 解压到目录 /usr/local/zookeeper。

创建数据目录并准备配置文件：

mkdir /usr/local/zookeeper/tmp
cp /usr/local/zookeeper/conf/zoo_sample.cfg /usr/local/zookeeper/conf/zoo.cfg
vim /usr/local/zookeeper/conf/zoo.cfg

在 zoo.cfg 中设置数据目录：

dataDir=/usr/local/zookeeper/tmp

启动 ZooKeeper 服务：

cd /usr/local/zookeeper/bin
zkServer.sh start

13. 流计算框架的安装部署与配置：Storm（单机）

将 apache-storm-0.9.6.tar.gz 解压到目录 /usr/local/storm。

编辑配置文件 /usr/local/storm/conf/storm.yaml：

storm.zookeeper.servers:
    - "127.0.0.1"

nimbus.host: "nimbus"

简单配置后即可启动 Storm。

13.1 启动 nimbus

cd /usr/local/storm/bin/
storm nimbus

启动 nimbus 后，终端会被该进程占用，不能继续执行其他命令。因此需要另外开启一个终端，执行如下命令启动 supervisor 后台进程：

/usr/local/storm/bin/storm supervisor

同样，启动 supervisor 后终端也会被占用，需要再开启其他终端才能执行命令。

13.2 检查启动状态

可以使用 jps 命令检查是否成功启动。若成功启动会显示 nimbus、supervisor、QuorumPeerMain（QuorumPeerMain 是 ZooKeeper 的后台进程；若显示 config_value 表明 nimbus 或 supervisor 还在启动中）：

jps

3345 DataNode
5953 supervisor
5827 nimbus
3204 NameNode
3532 SecondaryNameNode
5630 QuorumPeerMain
6063 Jps

13.3 关闭 Storm

之前启动的 nimbus 和 supervisor 占用了两个终端窗口。切换到这两个终端窗口，按键盘的 Ctrl+C 即可终止进程，终止后也就相当于关闭了 Storm。

14. 提交与运行流计算作业：WordCount 实例

Storm 自带了一些例子，可以执行 WordCount 例子来感受 Storm 的执行流程。执行如下命令：

/usr/local/storm/bin/storm jar /usr/local/storm/examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology

该程序不断地从如下几句英文句子中取一句作为数据源，然后发送给 bolt 来统计单词出现的次数：

the cow jumped over the moon
an apple a day keeps the doctor away
four score and seven years ago
snow white and the seven dwarfs
i am at two with nature

该程序执行 10 秒后会自动关闭。执行过程中会输出非常多的信息（info 级别的日志），可以使用 grep 命令过滤输出，只显示关心的单词实时统计信息（只显示用于统计单词次数的 bolt 的输出信息）：

/usr/local/storm/bin/storm jar /usr/local/storm/examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology | grep 'Thread-[0-9]*-count'

更具体的执行流程可以查看 WordCount 的代码，文件位置为：

/usr/local/storm/examples/storm-starter/src/jvm/storm/starter/WordCountTopology.java

Storm 官方提供了一个很好的入门代码：Storm starter。

15. 结果验证

一次处理会输出两条信息：一是显示接收到的数据（单词 ["day"]），二是输出该单词当前总的出现次数（[day, 30]，表示当前 "day" 已出现了 30 次）。

由此可见流计算结果的特征：结果是持续更新的，同一个单词的计数会随着数据流的持续输入而不断增大，而不像批处理那样只在作业结束时给出一个固定结果。

16. 使用 Maven 编译运行 Storm starter

16.1 Maven 介绍

Maven 是现在 Java 社区中最强大的项目管理和项目构建工具。开发项目过程中会使用一些开源框架、第三方工具等，这些都以 jar 包的方式被项目引用，并且有些 jar 包还会依赖其他 jar 包，所有这些相关 jar 包都会作为项目的依赖。通常一个 Java EE 项目依赖的 jar 包会很多，但这并非主要问题——在管理这些 jar 包的过程中，jar 包的版本往往最令人头疼：选择一个 jar 包的版本，需要考虑它依赖的 jar 包是否支持该版本，还要确认依赖它的 jar 包能否兼容该版本，难以部署与管理。

Maven 的出现解决了这一难题。它可以对项目依赖的 jar 包进行管理，让项目保持基本的依赖、排除冗余 jar 包，并能轻松对依赖的 jar 包进行版本升级。这些仅是 Maven 最基本的功能，在此基础上它还可以对项目进行清理、编译、测试、打包、发布等构建工作。

Maven 通过 pom.xml（Project Object Model，项目对象模型）来对项目进行配置。可以查看 storm-starter 目录下的 pom.xml 来了解运行该入门代码所需的各种 jar 包。

16.2 安装 Maven

将 apache-maven-3.3.9-bin.tar 解压到 /usr/local/maven：

cd ~
tar -xvf apache-maven-3.3.9-bin.tar
mv apache-maven-3.3.9 /usr/local/maven

配置环境变量，编辑 /etc/profile，在尾部添加：

export PATH=/usr/local/maven/bin:$PATH

使配置生效并验证：

source /etc/profile
mvn -v

Apache Maven 3.3.9 (bb52d8502b132ec0a5a3f4c09453c07478323dc5; 2015-11-10T08:41:47-08:00)
Maven home: /usr/local/maven
Java version: 1.8.0_191, vendor: Oracle Corporation
Java home: /usr/local/jdk/jre
Default locale: zh_CN, platform encoding: UTF-8
OS name: "linux", version: "3.10.0-514.el7.x86_64", arch: "amd64", family: "unix"

16.3 编译、运行代码

在此之前请先开启 Storm（即开启 ZooKeeper、nimbus 和 supervisor），并确保系统处于联网状态，因为编译过程中需要下载不少依赖包。

进入 storm-starter 所在目录，执行相应的 mvn 命令并指定参数。以运行 WordCount 为例（源码位于 /usr/local/storm/examples/storm-starter 下的 src/jvm/storm/starter/WordCountTopology.java）：

cd /usr/local/storm/examples/storm-starter
mvn compile exec:java -Dstorm.topology=storm.starter.WordCountTopology

mvn 命令需要在项目根目录（即 storm-starter 目录）中运行，它会查找当前目录下的 pom.xml 文件并下载所需依赖包。首次运行需要下载较多内容，请耐心等待。

再次运行并过滤出单词统计信息：

mvn compile exec:java -Dstorm.topology=storm.starter.WordCountTopology | grep 'Thread-[0-9]*-count'

此时不需要再下载大量内容，编译很快完成。只输出单词统计信息，可以看到结果与直接运行 Storm 安装实例时一致，说明编译正确。

16.4 打包代码

也可以将代码打包成 jar，打包后可直接提交到 Storm 中运行。在 storm-starter 所在目录执行如下命令（这样打包的是整个 Storm starter）：

mvn package

打包完成后，会在 target 目录中生成 storm-starter-0.9.6.jar 和 storm-starter-0.9.6-with-dependencies.jar，就可以把 jar 包提交到 Storm 中运行了：

/usr/local/storm/bin/storm jar ./target/storm-starter-0.9.6.jar storm.starter.WordCountTopology

17. 小结与常见问题

17.1 小结

后续编写自己的代码时，建议使用 Maven 来方便地编译、运行代码。官方使用说明中有更详细的介绍，例如在 IntelliJ 中运行 Storm starter，有需要可自行查看。

Storm starter 提供了很好的示例代码（源代码位于 storm-starter/src/jvm/storm/starter/ 中），例如 RollingTopWords，可为自己编写 Storm 代码提供很好的借鉴。

17.2 常见问题

18. 常用命令速查

练习

2. 说明流计算的三个处理阶段，并在 Storm WordCount 实例中分别指出对应的组件（Spout / Bolt / 输出）。

3. 对比 Storm、Spark Streaming、Flink、Samza 四种框架的处理模型、延迟级别与语义保证，并说明各自最适合的场景。

4. 完成 ZooKeeper 单机部署（创建 tmp 目录、由 zoo_sample.cfg 生成 zoo.cfg、设置 dataDir）并启动，用 jps 确认 QuorumPeerMain 进程存在。

5. 完成 Storm 单机安装：配置 storm.yaml 的 storm.zookeeper.servers 与 nimbus.host，分别在两个终端启动 nimbus 与 supervisor，用 jps 验证。

6. 提交并运行官方 WordCount 拓扑，用 grep 'Thread-[0-9]*-count' 过滤输出，解释输出的两类信息分别代表什么。

7. 阅读 /usr/local/storm/examples/storm-starter/src/jvm/storm/starter/WordCountTopology.java，画出该拓扑的 Spout/Bolt 结构图。

8. 安装 Maven 并配置环境变量，用 mvn -v 验证；然后用 mvn compile exec:java 运行 WordCountTopology，并与直接运行安装实例的结果做对比。

9. 用 mvn package 打包 storm-starter，说明生成的 storm-starter-0.9.6.jar 与 storm-starter-0.9.6-with-dependencies.jar 的区别，并把前者提交到 Storm 运行。

10. 阅读 storm-starter 中的 RollingTopWords 示例，说明它实现的"滑动窗口热词统计"与 WordCount 的累计统计有何不同。

11. 查看 storm-starter 目录下的 pom.xml，列出运行该入门代码所依赖的主要 jar 包及其作用。