Storm命令详解-Storm+kafka的HelloWorld初体验

忘是亡心i 2022-04-04 07:17 521阅读 0赞

# Storm命令详解 #

2015年04月26日 15:08:08 [lavimer][] 阅读数：9171 标签： [Storm ][Storm][Storm命令详解 ][Storm_][Storm命令 ][Storm_ 1] 更多

个人分类： [Storm ][Storm 1]

所属专栏： [Storm实战][Storm 2]

在Linux终端直接输入storm，不带任何参数信息，或者输入storm help，可以查看storm命令行客户端(Command line client)提供的帮助信息。Storm 0.9.0.1版本在Linux终端直接输入storm后的输出内容如下：

1.  `Commands:`
2.  `activate`
3.  `classpath`
4.  `deactivate`
5.  `dev-zookeeper`
6.  `drpc`
7.  `help`
8.  `jar`
9.  `kill`
10. `list`
11. `localconfvalue`
12. `logviewer`
13. `nimbus`
14. `rebalance`
15. `remoteconfvalue`
16. `repl`
17. `shell`
18. `supervisor`
19. `ui`
20. `version`
21.  
22. `Help:`
23. `help`
24. `help <command>`
25.  
26. `Documentation for the storm client can be found at https://github.com/nathanmarz/storm/wiki/Command-line-client`
27.  
28. `Configs can be overridden using one or more -c flags, e.g. "storm list -c nimbus.host=nimbus.mycompany.com"`

**注：由此可知，新版Storm的命令行客户端提供了19个命令。**

**1.activate**

激活指定的拓扑。语法如下：

storm activate topology-name

**2.classpath**

打印出Storm客户端运行命令时使用的类路径(classpath)。语法如下：

storm classpath

**3.deactivate**

禁用指定的拓扑Spout。语法如下：

storm deactivate topology-name

**4.dev-zookeeper**

以dev.zookeeper.path配置的值作为本地目录，以storm.zookeeper.port配置的值作为端口，启动一个新的Zookeeper服务，仅用来开发/测试。语法如下：

storm dev-zookeeper

**5.drpc (常用！)**

启动一个DRPC守护进程。语法如下：

storm drpc

**注：该命令应该使用daemontools或者monit工具监控运行。**

**6.help (常用！)**

打印一条帮助消息或者可用命令的列表。语法如下：

1.  `storm help`
2.  `storm help <command>`

**注：直接输入不带参数的storm，也可以启动storm help命令。**

**7.jar (很常用！)**

运行类的指定参数的main方法。语法如下：

storm jar topology-jar-path class ...

**注：把Storm的jar文件和"~/.storm"的配置放到类路径(classpath)中，以便当拓扑提交时，StormSUbmitter会上传topology-jar-path的jar文件。**

**8.kill (常用！)**

杀死名为topology-name的拓扑。语法如下：

storm kill topology-name [-w wait-time-secs]

**注：storm首先会在拓扑的消息超时时间期间禁用spout，以允许所有正在处理的消息完成。然后，Storm将会关闭Worker并清理他们的状态。可以使用-w标记覆盖Storm在禁用与关闭期间等待的时间长度。**

**9.list (常用！)**

列出正在运行的拓扑及其状态。语法如下：

storm list

**10.localconfvalue**

打印出本地Storm配置的conf-name的值。语法如下：

storm localconfvalue conf-name

**注：本地Storm配置是~/.storm/storm.yaml与defaults.yaml合并的结果。**

**11.logviewer (常用！)**

启动Logviewer守护进程。语法如下：

storm logviewer

**注：Logviewer提供一个Web接口查看Storm日志文件。该命令应该使用daemontools或者monit工具监控运行。**

**12.nimbus (常用！)**

启动Nimbus守护进程。语法如下：

storm nimbus

**注：该命令应该使用daemontools或者monit工具监控运行。**

**13.rebalance (常用！)**

再平衡即动态设置拓扑的进程数量和线程数量等。详细内容见：[这里][Link 1]

**14.remoteconfvalue**

打印出远程集群Storm配置的conf-name的值。语法如下：

storm remoteconfvalue conf-name

**注：集群Storm配置是$STORM-PATH/conf/storm.yaml与defaults.yaml合并的结果。该命令必须在集群节点上运行。**

**15.repl**

打开一个包含路径(classpath)中的jar文件和配置的Clojure REPL，以便调试时使用。语法如下：

storm repl

**注：Clojure可以作为一种脚本语言内嵌到java中，但是Clojure的首选编程方式是使用REPL，REPL是一个简单的命令行接口。使用REPL，可以输入命令并执行，然后查看结果。**

**16.shell**

执行Shell脚本。语法如下：

storm shell resourcesdir command args

**17.supervisor (常用！)**

启动Supervisor守护进程。语法如下：

storm supervisor

**注：该命令应该使用daemontools或者monit工具监控运行。**

**18.ui (常用！)**

启动UI守护进程。语法如下：

storm ui

**注：UI为Storm集群提供了一个Web界面并显示运行拓扑的详细统计信息。该命令应该使用daemontools或者monit工具监控运行。**

**19.version**

打印Storm发布的版本号。语法如下：

storm version

**附：文章引用自《从零开始学Storm》**

# [Storm+kafka的HelloWorld初体验][Storm_kafka_HelloWorld] #

从16年4月5号开始学习kafka，后来由于项目需要又涉及到了storm。

经过几天的扫盲，到今天16年4月13日，磕磕碰碰的总算是写了一个kafka+storm的HelloWorld的例子。

为了达到前人栽树后人乘凉的知识共享的目的，我尝试着梳理一下过程。

**====实例需求**

由kafka消息队列源源不断生产数据，然后由storm进行实时消费。

大家可以设想这些数据源是不同商品的用户行为操作行为，我们是不是就可以实时观测到用户关注商品的热点呢？

**====环境准备**

（1）Linux：

公司暂时没有多余的Linux主机，所以我只能在自己的电脑上建立的3台Linux虚拟机。

虚拟机的建立方法我做了一个小白级别的手册，按照这个手册就可以建立起虚拟机了。

百度云连接地址：[http://pan.baidu.com/s/1hr3lVqG][http_pan.baidu.com_s_1hr3lVqG]

（2）JDK：

我这里使用的是：jdk-7u80-linux-x64.tar.gz。

在官方网站上下载，然后配置环境变量即可。

（3）zookeeper集群：

搭建方法省略。可以参照我的博客：[http://www.cnblogs.com/quchunhui/p/5356511.html][http_www.cnblogs.com_quchunhui_p_5356511.html]

（4）kafka：

搭建方法省略。可以参照我的博客：[http://www.cnblogs.com/quchunhui/p/5356511.html][http_www.cnblogs.com_quchunhui_p_5356511.html]

（5）storm：

我这里使用的版本是相对稳定的：apache-storm-0.9.5.tar.gz

搭建方法省略，可以参照我的博客：[http://www.cnblogs.com/quchunhui/p/5370191.html][http_www.cnblogs.com_quchunhui_p_5370191.html]

（6）Maven：

开发环境的构建使用Maven。我这里使用的版本是：apache-maven-3.3.3.zip

Maven的入门可以参考我的博客：[http://www.cnblogs.com/quchunhui/p/5359293.html][http_www.cnblogs.com_quchunhui_p_5359293.html]

补充一下环境变量配置之后的图，以供小白参考。

![915691-20160413225741598-1535251411.png][]

**====程序执行方式**

（1）kafka：

需要手动编写kafka的生产者程序，然后通过eclipse等工具在Windows端启动，以达到生产消息的目的。

（2）storm：

可以进行两种方式的启动。一种是通过eclipse等工具在Windows端启动（俗称本地模式）

另一种是将storm的消费者程序打成jar包发布到Linux环境上，通过Linux启动程序进行消费（俗称集群模式）。

**====Storm框架前期理解**

从某位大神的QQ群组里下载了一篇关于storm的基本框架以及安装的文章

我这里共享到了我的百度云盘上了，请大家在开始编程之前一定要看看。非常值得一看。

百度云地址：[http://pan.baidu.com/s/1boRcCeb][http_pan.baidu.com_s_1boRcCeb]

那么后面我们就可以开始编写我们的程序了。首先需要编写的是kafka的生产者程序。

**====kafka程序相关：**

我已经写好的代码共享到了Github上了：[https://github.com/quchunhui/kafkaSample/][https_github.com_quchunhui_kafkaSample]

这里只对目录结构以及重要部分进行说明：

（1）src/main路径结构如下：

\+---common

（2）实例所用的代码：

本次实例中，仅仅使用了kafka进行消息的生产，同事考虑到异步生产性能更高一些，

本次实例中使用了异步生产的代码，就是上面红色字标记的java程序（AsyncProduce.java）。

代码本身比较简单，其中下面红色框的部分为【异步】的配置项，需要注意。

各个配置项的说明请参考我的另一篇博客：[http://www.cnblogs.com/quchunhui/p/5357040.html][http_www.cnblogs.com_quchunhui_p_5357040.html]

![915691-20160413232115129-1521193122.png][]

**====Storm程序相关：**

（1）拓扑设计

![915691-20160411214005535-1452289335.png][]

【消息源（RandomSentenceSpout）】

接入到从上面的kafka消息队列中，将kafka作为storm的消息源。

【数据标准化（WordNormalizerBolt）】

然后使用一个Bolt进行归一化（语句切分），句子切分成单词发射出去。（代码更新中。。。）

【词频统计（WordCountBolt）】

使用一个Bolt接受订阅切分的单词Tuple，进行单词统计，并且选择使用按字段分组的策略，词频实时排序，把TopN实时发射出去。（代码更新中。。。）

【工具类（PrintBolt）】

最后使用一哥Bolt将结果打印到Log中。（代码更新中。。。）

**====实例代码**

我自己进行验证用的代码已经上传到Github上了，可以直接下载下来使用。

这里只对代码的目录结构以及需要格外关注的点进行一些补充。

Git地址：[https://github.com/quchunhui/storm-kafka-plus-qch][https_github.com_quchunhui_storm-kafka-plus-qch]

（1）目录结构

src\\main\\java\\com\\dscn\\helloworld

|   
| WordCountTopology.java         // Topology代码，程序入口，使用eclipse是需要执行该程序。  
|  
\+---bolt  
| PrintBolt.java                          // 上面讲到的工具类（PrintBolt）类  
| WordCountBolt.java                // 上面讲到的词频统计（WordCountBolt）类  
| WordNormalizerBolt.java          // 上面讲到的数据标准化（WordNormalizerBolt）类  
|  
\\---spout  
RandomSentenceSpout.java       // 未使用

（2）重要代码说明

由于源代码已经共享给大家了，Storm的接口的用法在下面的篇幅中单独罗列了一下，我这里不进行过多的阐述。

在这里只将我碰到过的问题罗列出来、以问题&解决方法的形式分享。

【问题1】

storm是如何实现与kafka的对接的

【回答】

Spout作为storm的消息源的接入点，在这里可以同构设置Storm官方提供【storm.kafka.SpoutConfig】类来指定消息源。

![915691-20160413234435395-1547957938.png][]

\----------------

//配置zookeeper集群地址，毕竟storm是需要集群支持的。

BrokerHosts brokerHosts = new ZkHosts("192.168.93.128:2181,192.168.93.129:2181,192.168.93.130:2181");

//配置Kafka订阅的Topic，以及zookeeper中数据节点目录和名字

SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, "qchlocaltest", "", "topo");

spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

//如果该Topology因故障停止处理，下次正常运行时是否从Spout对应数据源Kafka中的该订阅Topic的起始位置开始读取  
spoutConfig.forceFromStart = true;

//zookeeper集群host

spoutConfig.zkServers = Arrays.asList(new String\[\] \{"192.168.93.128", "192.168.93.129", "192.168.93.130"\});

//zookeeper集群port

spoutConfig.zkPort = 2181;

\----------------

【问题2】

我尝试着自己重新写代码配置开发环境（不是直接使用Github上的代码），

编译时可以正常通过的，但是本地模式通过eclipse启动Topology的时候，出现了log4j和slf4j的冲突问题。

【解决方法】

问题原因是由于log4j和slf4j之间的重复调用，导致死循环而致使内存溢出。

解决办法就是log4j和slf4j保留一个， 普遍上都是保留slf4j的。

需要在Maven的pom.xml上将log4j的相关依赖移除。

移除方法：

![915691-20160414000606910-931678189.png][]

可以通过【mvn dependency:tree】命令来查看修改之后的依赖关系。

如果发现需要移除的包的时候，使用Maven的【exclusion】标签来移除依赖关系。

填写exclusion标签的时候，下图中红色的部分是groupId，蓝色的部分是artifactId。

![915691-20160414001145316-1602106043.png][]

【问题3】

使用mvn install命令将程序打jar包上传到Linux的storm目录下，然后使用命令

\[storm jar test-0.1-jar-with-dependencies.jar com.dscn.helloworld.WordCountTopology 192.168.93.128\]

启动Topology的时候，出现了下面的提示错误。

![915691-20160414002022754-591154931.png][]

【解决方法】

是Maven的pom.xml的配置出现了问题。详细请参考博客：[http://blog.csdn.net/luyee2010/article/details/18455237][http_blog.csdn.net_luyee2010_article_details_18455237]

修改方法就是强storm的scope修改为provided。如下图所示：

![915691-20160414002422113-780267606.png][]

【问题4】

将代码放到实际的集群运行环境（kafka+storm+hbase）中，发现storm接受不到消息。

【原因】

一直以来都是使用kafka的异步生产来生产消息，以为都正常的生产消息了。由于异步生产的时候，并没有消息确认机制，

所以不能确保消息是否正确的进入到了消息队列之中，改用同步生产的代码尝试了一下，果然发生了一下的错误。

![915691-20160419113011726-1034087454.png][]

【解决办法】

通过网上搜索\[kafka Failed to send messages\]关键字，发现有可能是需要设置advertised.host.name这个属性。

抱着尝试一下的心态试了一下，果然好使了。至于这个属性的真正意义还有待探索。**（TODO）**

![915691-20160419113157070-1129245854.png][]

【问题5】

代码在本地的时候好好的，通过storm jar命令发布到集群环境的时候，发生了Jar包冲突的问题。

![915691-20160419114054507-2026737821.png][]

【解决方法】

本来是认为自己的Maven环境的依赖有问题，也通过mvn dependency:tree查看了依赖关系，毫无问题。根本就诶有log4j-over-slf4j.jar这个包。

头疼了很久，通过QQ群咨询了一些朋友，他们建议我确认集群环境中storm/lib下是否存在log4j-over-slf4j.jar，如果存在就把它删掉。

尝试了一下之后，果然好使了。原来是我的程序的jar包和集群环境中会有冲突。详细请参考我的另一篇博客：

[http://www.cnblogs.com/quchunhui/p/5404168.html][http_www.cnblogs.com_quchunhui_p_5404168.html]

**====Storm接口详解：**

【IComponent接口】

Spout和Bolt都是其Component。所以，Storm定义了一个名叫IComponent的总接口。

![915691-20160411205855738-509255293.png][]

IComponent的继承关系如下图所示：

![915691-20160411205628098-479110359.png][]

绿色部分是我们最常用、比较简单的部分。红色部分是与事务相关。

BaseComponent 是Storm提供的“偷懒”的类。为什么这么说呢，它及其子类，都或多或少实现了其接口定义的部分方法。

这样我们在用的时候，可以直接继承该类，而不是自己每次都写所有的方法。

但值得一提的是，BaseXXX这种定义的类，它所实现的方法，都是空的，直接返回null。

**【Spout】**

类图如下图所示：

![915691-20160411210240238-1020379653.png][]

接口如下图所示：

![915691-20160411210538285-647803027.png][]

各个接口说明：

①、open方法：

是初始化动作。允许你在该spout初始化时做一些动作，传入了上下文，方便取上下文的一些数据。

②、close方法

在该spout关闭前执行，但是并不能得到保证其一定被执行。

spout是作为task运行在worker内，在cluster模式下，supervisor会直接kill -9 woker的进程，这样它就无法执行了。

而在本地模式下，只要不是kill -9, 如果是发送停止命令，是可以保证close的执行的。

③、activate和deactivate方法 ：

一个spout可以被暂时激活和关闭，这两个方法分别在对应的时刻被调用。

④、nextTuple方法：

负责消息的接入，执行数据发射。是Spout中的最重要方法。

⑤、ack(Object)方法：

传入的Object其实是一个id，唯一表示一个tuple。该方法是这个id所对应的tuple被成功处理后执行。

⑥、fail(Object)方法：

同ack，只不过是tuple处理失败时执行。

我们的RandomSpout由于继承了BaseRichSpout，

所以不用实现close、activate、deactivate、ack、fail和getComponentConfiguration方法，只关心最基本核心的部分。

**结论：**

通常情况下（Shell和事务型的除外），实现一个Spout，可以直接实现接口IRichSpout，如果不想写多余的代码，可以直接继承BaseRichSpout。

**【Bolt】**

类图如下图所示：

![915691-20160411211450707-1414793197.png][]

这里可以看到一个奇怪的问题： 为什么IBasicBolt并没有继承IBolt？ 我们带着问题往下看。

IBolt定义了三个方法：

![915691-20160411211848160-1844366873.png][]

①、prepare方法：

IBolt继承了java.io.Serializable，我们在nimbus上提交了topology以后，创建出来的bolt会序列化后发送到具体执行的worker上去。

worker在执行该Bolt时，会先调用prepare方法传入当前执行的上下文。

②、execute方法：

接受一个tuple进行处理，并用prepare方法传入的OutputCollector的ack方法（表示成功）或fail（表示失败）来反馈处理结果。

③、cleanup方法：

同ISpout的close方法，在关闭前调用。同样不保证其一定执行。

红色部分（execute方法）是Bolt实现时一定要注意的地方。

而Storm提供了IBasicBolt接口，其目的就是实现该接口的Bolt不用在代码中提供反馈结果了，Storm内部会自动反馈成功。

如果你确实要反馈失败，可以抛出FailedException。

我们来再写一个Bolt继承BaseRichBolt替代ExclaimBasicBolt。代码如下：

![915691-20160411212841691-1974318975.png][]

修改topology

![915691-20160411212926473-1202064173.png][]

运行下，结果一致。

**结论：**

通常情况下，实现一个Bolt，可以实现IRichBolt接口或继承BaseRichBolt，

如果不想自己处理结果反馈，可以实现IBasicBolt接口或继承BaseBasicBolt，它实际上相当于自动做掉了prepare方法和collector.emit.ack(inputTuple)；

**====推荐博客：**

【整合实战类】：

http://shiyanjun.cn/archives/934.html

http://www.tuicool.com/articles/NzyqAn

http://itindex.net/detail/51477-storm-笔记-kafka

【问题解决类】：

http://www.aboutyun.com/thread-12590-1-1.html

【Storm调优类】：

http://blog.csdn.net/derekjiang/article/details/9040243

http://www.51studyit.com/html/notes/20140329/45.html

\--END--

[lavimer]: https://me.csdn.net/lzm1340458776
[Storm]: http://so.csdn.net/so/search/s.do?q=Storm&t=blog
[Storm_]: http://so.csdn.net/so/search/s.do?q=Storm%E5%91%BD%E4%BB%A4%E8%AF%A6%E8%A7%A3&t=blog
[Storm_ 1]: http://so.csdn.net/so/search/s.do?q=Storm%E5%91%BD%E4%BB%A4&t=blog
[Storm 1]: https://blog.csdn.net/lzm1340458776/article/category/3095395
[Storm 2]: https://blog.csdn.net/column/details/storm-lavimer.html
[Link 1]: http://blog.csdn.net/lzm1340458776/article/details/45286207
[Storm_kafka_HelloWorld]: https://www.cnblogs.com/quchunhui/p/5380260.html
[http_pan.baidu.com_s_1hr3lVqG]: http://pan.baidu.com/s/1hr3lVqG
[http_www.cnblogs.com_quchunhui_p_5356511.html]: http://www.cnblogs.com/quchunhui/p/5356511.html
[http_www.cnblogs.com_quchunhui_p_5370191.html]: http://www.cnblogs.com/quchunhui/p/5370191.html
[http_www.cnblogs.com_quchunhui_p_5359293.html]: http://www.cnblogs.com/quchunhui/p/5359293.html
[915691-20160413225741598-1535251411.png]: /images/20220404/8196e7c7ee8349ce9df684ebb1e822a3.png
[http_pan.baidu.com_s_1boRcCeb]: http://pan.baidu.com/s/1boRcCeb
[https_github.com_quchunhui_kafkaSample]: https://github.com/quchunhui/kafkaSample/
[http_www.cnblogs.com_quchunhui_p_5357040.html]: http://www.cnblogs.com/quchunhui/p/5357040.html
[915691-20160413232115129-1521193122.png]: /images/20220404/08b56debb8ca4c04badb9bbfc6a119c1.png
[915691-20160411214005535-1452289335.png]: /images/20220404/39553226f0f54eb68f030bfad44212c7.png
[https_github.com_quchunhui_storm-kafka-plus-qch]: https://github.com/quchunhui/storm-kafka-plus-qch
[915691-20160413234435395-1547957938.png]: /images/20220404/b81c40e434c9430eac7322a5de8616a7.png
[915691-20160414000606910-931678189.png]: /images/20220404/9be78c21afab412e8846bde196ccd411.png
[915691-20160414001145316-1602106043.png]: /images/20220404/ce7b5da347cd44c792ecd9ab23828517.png
[915691-20160414002022754-591154931.png]: /images/20220404/f83bad5a2dd3473ead61992b4d8bde2a.png
[http_blog.csdn.net_luyee2010_article_details_18455237]: http://blog.csdn.net/luyee2010/article/details/18455237
[915691-20160414002422113-780267606.png]: /images/20220404/80ead5d39fa64bec8d11e338d18c0e7d.png
[915691-20160419113011726-1034087454.png]: /images/20220404/e42ab6f4f27146f68d2824b0b1495579.png
[915691-20160419113157070-1129245854.png]: /images/20220404/d7e6d9bdca4b423bb8b31d9697fa54d6.png
[915691-20160419114054507-2026737821.png]: /images/20220404/d5b8b06d26fa4d819868a8933ce373da.png
[http_www.cnblogs.com_quchunhui_p_5404168.html]: http://www.cnblogs.com/quchunhui/p/5404168.html
[915691-20160411205855738-509255293.png]: /images/20220404/01384bf5a1ff42d1adf7ec9b694d30f5.png
[915691-20160411205628098-479110359.png]: /images/20220404/76eeb8f0d583463a9c59fa803dc59f73.png
[915691-20160411210240238-1020379653.png]: /images/20220404/49eadc4695024e4f82bd91c8906be61d.png
[915691-20160411210538285-647803027.png]: /images/20220404/35652c97fad642ba99bad4b50148ecc4.png
[915691-20160411211450707-1414793197.png]: /images/20220404/bd4602f5de7447ba899e99b9e0cba0c9.png
[915691-20160411211848160-1844366873.png]: /images/20220404/b78af91a31914a8aafdb0e99ddf2c626.png
[915691-20160411212841691-1974318975.png]: /images/20220404/68eb4122541e478c9897c1d81f54b24e.png
[915691-20160411212926473-1202064173.png]: /images/20220404/d174d0d8bbe043d38a0234431f9084c5.png