Storm框架基础（一）

以你之姓@ 2022-05-19 08:52 170阅读 0赞

来源:[https://www.jianshu.com/p/7e5fc624861b][https_www.jianshu.com_p_7e5fc624861b]

## Storm简述 ##

如果你了解过SparkStreaming，那么Storm就可以类比着入门，在此我们可以先做一个简单的比较：

** 在SparkStreaming中：**

我们曾尝试过每秒钟的实时数据处理，或者使用Window若干时间范围内的数据统一处理结果。亦或统计所有时间范围内的数据结果。

**在Storm中：**

我们可以根据进来的每一条数据进行实时处理，也就是说，Storm处理数据的速度，要小于1秒，也就是毫秒级别的。

如果你疑问，1秒处理1次数据，和进来1条数据处理1次有什么差别，稍作思考答案应该是显而易见的，因为1秒中就可以产生不计其数的数据。

## Storm角色 ##

主节点：Nimbus

从节点：Supervisor

## Storm基本概念 ##

**Nimbus：**主节点

**Supervisor：**从节点

**Worker：**任务工作进程，类似于YARN的ApplicationMaster，可以存在多个，不同的任务有不同的Worker

**Executor：**Worker进程在执行任务时，会启动多个Executor现成

**Topology：**task任务的拓扑结构，是一个DAG有向无环图

**Spout：**抽取数据并将数据进行分发的阶段

**Bolt：**将分发的数据进行具体操作的阶段

## Storm集群配置 ##

![Image 1][]

zookeeper在这里起到同步主/从节点状态信息的作用，后续会详解，所以Storm是依赖zookeeper的

## Storm的安装部署 ##

OK，闲话不多说，我们直接开始部署Storm，部署完成后，我们再根据运行的任务，来解析一下Storm任务运行的流程和调度过程以及其中的简单原理。

本次安装需要依赖Java以及Python环境，如果当前虚拟机没有，请自行安装，安装后检测，如图：

![Image 1][]

注意Python是2版本系列

官网网站：storm.apache.org

Storm下载传送门：

链接：http://pan.baidu.com/s/1pLwj4dt 密码：resv

**Step1、解压**

$ tar -zxf apache-storm-0.9.6.tar.gz -C  /opt/modules/

**Step2、修改配置文件**

**尖叫提示：Notepad++编辑配置文件可能会导致yaml文件格式错乱，请留意，建议使用sublime或者vi编辑**

**storm\_env.ini**

![Image 1][]

**storm.yaml**

配置帮助页面：http://storm.apache.org/releases/0.9.6/Setting-up-a-Storm-cluster.html

![Image 1][]

配置完成后将安装包scp到其他机器节点。

**Step3、启动三台节点的zookeeper集群**

$ /opt/modules/cdh/zookeeper-3.4.5-cdh5.3.6/bin/zkServer.sh start

**Step4、启动Nimbus和UI进程**

$ nohup bin/storm nimbus > /dev/null 2>&1 &

$ nohup bin/storm ui > /dev/null 2>&1 &

**检查：**

$ jps

![Image 1][]

启动后需要稍等片刻，才能config\_value从这个配置进程切换到对应进程

$ ps -ef | grep daemon.nimbus

$ ps -ef | grep ui.core

如图：

![Image 1][]

**Step5、所有节点启动Supervisor和logviewer进程**

$ nohup bin/storm supervisor > /dev/null 2>&1 &

$ nohup bin/storm logviewer > /dev/null 2>&1 &

**检查第一台节点（第二，第三台节点请自行检查）：**

$ jps

![Image 1][]

$ ps -ef | grep daemon.supervisor

$ ps -ef | grep daemon.logviewer

![Image 1][]

**尖叫提示：**

默认情况下，nohup执行的日志在当前目录下的nohup.out中，但此刻我们将日志重定向到了/dev/null设备中（即空设备）意思就是丢弃不要了。

Step6、登录z01:8081查看storm集群启动情况，如下图所示

![Image 1][]

**Step7、运行个官方案例，测试一下吧**

$ bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount

任务提交后，可以在此查看：

![Image 1][]

打开该任务，可以看到有一个DAG图：

![Image 1][]

后续涉及到代码编写的时候自然会明白该图在说什么

## **Storm集群运行机制** ##

**主节点Nimbus**

管理所有从节点supervisor，当主节点nimbus启动后会定时向zookeeper集群发送自己的当前状态信息，同时也可以获取所有从节点Supervisor的状态信息

**从节点Supervisor**

定时向zookeeper集群汇报自己的状态信息，同时接受主节点Nimbus派发过来的任务。

**Zookeeper的作用**

Storm的所有节点的状态信息都保存在zookeeper当中，如果有某个节点挂掉了，只需要启动一个新的节点来替代即可，新的节点启动后，会自动从zookeeper中读取之前的状态信息，然后恢复到该状态下。同时也不必担心task任务意外终止，因为可以通过zookeeper来恢复该任务，也就是重启挂掉的task。nimbus和supervisor之间的状态同步全部依赖zookeeper来实现，看如下一张图，该图是zookeepr中storm的目录结构：

$ /opt/modules/cdh/zookeeper-3.4.5-cdh5.3.6/bin/zkCli.sh -server z01:2181

\[zk: z01:2181(CONNECTED) 0\] ls /

\[hbase, hadoop-ha, admin, zookeeper, consumers, config, rmstore, storm, yarn-leader-election, brokers, controller\_epoch\]

\[zk: z01:2181(CONNECTED) 1\] ls /storm

\[workerbeats, storms, supervisors, errors, assignments\]

![Image 1][]

**/workerbeats：**存储worker工作进程的运行状态信息

**/storms：**存储的Topology的基本配置信息

**/supervisors**：存储supervisor节点的状态信息

**/errors：**存储Topology任务运行过程中出现的异常信息

**/assignments：**存储的是Topology任务的分配信息

## Storm代码实现WordCount ##

**Step1、打开IDEA工具构建一个Maven工程，此步骤前文已经进行了多次练习，不再赘述**

![Image 1][]

**Step2、指定仓库地址，如果默认则无需改动**

![Image 1][]

**Step3、设置项目名称及工作路径，Finish即可**

![Image 1][]

**Step4、完事之后呢，打开pom.xml进行如下配置**

这个过程可能会有点慢，因为要从联网下载各种依赖包

![Image 1][]

**Step5、第四步全部搞定之后，如果下载了assembly插件，则需要添加如下配置，如果你不适用assembly则不需要Step5，你可以仿照前文Spark章节的打包过程进行。**

![Image 1][]

src.xml的内容是：

![Image 1][]

**Step6、我们就来根据这张图来编写代码**

![Image 1][]

这张图说明的是，storm的topology任务大概分为两个阶段，第一个阶段是spout阶段，主要负责数据的抽取，第二个阶段是bolt阶段，主要负责数据的处理，既然我们是要实现WordCount，所以Bolt数据处理阶段又分为两个过程，分别是单词的切分过程split和单词的汇总过程count，这两个过程都是bolt阶段。好，那么我们需要构建4个类，分别为：

WordCountTopology.Java

SentenceSpout.java

SplitBolt.java

CountBolt.java

如图：

![Image 1][]

**（尖叫提示：Storm中的Spout和Bolt类必须是可序列化的，****IDEA工具如何生成serialVersionUID请查看该文章：**[IDEA工具生成serialVersionUID][IDEA_serialVersionUID]）

**Step7、SentenceSpout.java实现细节**

![Image 1][]

**Step8、SplitBolt.java实现细节**

![Image 1][]

**Step9、CountBolt实现细节**

![Image 1][]

**Step10、WordCountTopology实现细节**

![Image 1][]

**Step11、激动人心的测试，Run起来之后，可以看到不停的有类似这样的结果输出**

![Image 1][]

**Step12、打包提交到集群测试**

打包时记得provided这个属性注释打开。

或者你使用IDEA自带的打包工具（请参考[SparkCore（二）][SparkCore]Spark打包部分内容）

总之你就是要弄个jar包出来，一切了然之后，我们先kill掉之前运行的wordcount，然后运行我们自己打出来的jar包。

$ bin/storm jar storm\_syllabus.jar com.z.WordCountTopology ZWordCount

![Image 1][]

UI界面查看：

![Image 1][]

注意，如果提交任务后，jps查看不到worker进程，也许是worker进程运行在其他节点，请依次查看。

so easy，right？

# \* 总结 #

学习完Spark后再学习Storm会简单很多，可以类比着来学习，其分布式运算的思想大多都有交集之处，慢慢体会。下一节我们继续探讨Storm框架。

第二节地址:

[https_www.jianshu.com_p_7e5fc624861b]: https://www.jianshu.com/p/7e5fc624861b
[Image 1]: 
[IDEA_serialVersionUID]: https://link.jianshu.com?t=http%3A%2F%2Fblog.csdn.net%2Fliuzongl2012%2Farticle%2Fdetails%2F45168585
[SparkCore]: https://www.jianshu.com/p/ec5b8869f2f4