Storm环境搭建和基本入门
程序员文章站
2022-06-19 19:50:18
...
要学习storm，我们可以先从并发编程网(ifeve.com)的storm入门开始：http://ifeve.com/getting-started-with-stom-index/
Storm集群安装部署
Storm 官方安装文档：http://storm.apache.org/releases/1.0.1/Setting-up-a-Storm-cluster.html，使用storm的版本1.0.1
zookeeper直接使用原有已经配置的即可，暂时不重新配置新的zookeeper集群, 测试一下zookeeper是否可以连接成功：
bash zkCli.sh -server 192.168.1.162:2181,192.168.1.163:2181,192.168.1.165:2181/solrcloud
连接成功后，显示：
2016-06-28 11:22:51,132 [myid:] - INFO [main-SendThread(192.168.1.163:2181):ClientCnxn$SendThread@1032] - Opening socket connection to server 192.168.1.163/192.168.1.163:2181. Will not attempt to authenticate using SASL (unknown error) 2016-06-28 11:22:51,143 [myid:] - INFO [main-SendThread(192.168.1.163:2181):ClientCnxn$SendThread@876] - Socket connection established to 192.168.1.163/192.168.1.163:2181, initiating session JLine support is enabled 2016-06-28 11:22:51,186 [myid:] - INFO [main-SendThread(192.168.1.163:2181):ClientCnxn$SendThread@1299] - Session establishment complete on server 192.168.1.163/192.168.1.163:2181, sessionid = 0x25484eb226700f2, negotiated timeout = 30000 [zk: 192.168.1.162:2181,192.168.1.163:2181,192.168.1.165:2181/solrcloud(CONNECTED) 0] WATCHER::
下载并解压storm后，修改 conf/storm.yaml文件，加入以下内容，用于指定storm的zookeeper服务端
storm.zookeeper.servers:
- “192.168.1.x"
- “192.168.1.x"
- “192.168.1.x"
配置storm本地存储目录，由于storm运行过程中，nimbus和supervisor守护进程需要在本地存储相应的jar包，配置文件等，需要指定 storm.local.dir=/usr/local/apache-storm-1.0.1/local/，如果不进行配置，根据文档中的说明，默认使用目录：
写道
If you run storm on windows,it could be: yaml storm.local.dir: "C:\\storm-local" If you use a relative path,it will be relative to where you installed storm(STORM_HOME). You can leave it empty with default value $STORM_HOME/storm-local
配置集群中配置nimbus的节点地址，各个supervisor工作节点需要知道哪个机器是nimbus，以便下载topologies的jars，confs等，nimbus.seeds:[“192.168.1.135”]，这里我们使用135这台服务器作为nimbus服务器地址。
supervisor.slots.ports: 对于每个Supervisor工作节点，需要配置该工作节点可以运行的worker数量。每个worker占用一个单独的端口用于接收消息，该配置选项即用于定义哪些端口是可被worker使用的。默认情况下，每个节点上可运行4个workers，分别在6700、6701、6702和6703端口，如：
supervisor.slots.ports:
- 6700
- 6701
- 6702
- 6703
这样在我们的每个supervisor中都设置了4个worker。
Storm启动过程
在配置完成之后，就可以启动storm集群了，storm集群的启动相对来说比较简单，通过几条命令就可以。
启动nimbus
bin/storm nimbus Running: /usr/java/jdk1.7.0_60/bin/java -server -Ddaemon.name=nimbus -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-1.0.1 -Dstorm.log.dir=/usr/local/apache-storm-1.0.1/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-1.0.1/lib/servlet-api-2.5.jar:/usr/local/apache-storm-1.0.1/lib/kryo-3.0.3.jar:/usr/local/apache-storm-1.0.1/lib/asm-5.0.3.jar:/usr/local/apache-storm-1.0.1/lib/objenesis-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-rename-hack-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-1.0.1/lib/disruptor-3.3.2.jar:/usr/local/apache-storm-1.0.1/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-1.0.1/lib/log4j-core-2.1.jar:/usr/local/apache-storm-1.0.1/lib/minlog-1.3.0.jar:/usr/local/apache-storm-1.0.1/lib/reflectasm-1.10.1.jar:/usr/local/apache-storm-1.0.1/lib/clojure-1.7.0.jar:/usr/local/apache-storm-1.0.1/lib/log4j-api-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-core-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-1.0.1/conf -Xmx1024m -Dlogfile.name=nimbus.log -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector -Dlog4j.configurationFile=/usr/local/apache-storm-1.0.1/log4j2/cluster.xml org.apache.storm.daemon.nimbus
启动supervisor
bin/storm supervisor Running: java -server -Ddaemon.name=supervisor -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-1.0.1 -Dstorm.log.dir=/usr/local/apache-storm-1.0.1/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-1.0.1/lib/servlet-api-2.5.jar:/usr/local/apache-storm-1.0.1/lib/kryo-3.0.3.jar:/usr/local/apache-storm-1.0.1/lib/asm-5.0.3.jar:/usr/local/apache-storm-1.0.1/lib/objenesis-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-rename-hack-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-1.0.1/lib/disruptor-3.3.2.jar:/usr/local/apache-storm-1.0.1/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-1.0.1/lib/log4j-core-2.1.jar:/usr/local/apache-storm-1.0.1/lib/minlog-1.3.0.jar:/usr/local/apache-storm-1.0.1/lib/reflectasm-1.10.1.jar:/usr/local/apache-storm-1.0.1/lib/clojure-1.7.0.jar:/usr/local/apache-storm-1.0.1/lib/log4j-api-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-core-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-1.0.1/conf -Xmx256m -Dlogfile.name=supervisor.log -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector -Dlog4j.configurationFile=/usr/local/apache-storm-1.0.1/log4j2/cluster.xml org.apache.storm.daemon.supervisor
开启ui监控
对于hadoop来说，其存在着对应的hadoop hdfs查看ui，以及任务执行web页面，相对hadoop来说storm的web ui略显简陋，但也能够以可视化的方式查看任务以及任务执行日志。
启动ui监控需要在nimbus所在服务器上执行 bin/storm ui命令：
Running: /usr/java/jdk1.7.0_60/bin/java -server -Ddaemon.name=ui -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-1.0.1 -Dstorm.log.dir=/usr/local/apache-storm-1.0.1/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-1.0.1/lib/servlet-api-2.5.jar:/usr/local/apache-storm-1.0.1/lib/kryo-3.0.3.jar:/usr/local/apache-storm-1.0.1/lib/asm-5.0.3.jar:/usr/local/apache-storm-1.0.1/lib/objenesis-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-rename-hack-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-1.0.1/lib/disruptor-3.3.2.jar:/usr/local/apache-storm-1.0.1/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-1.0.1/lib/log4j-core-2.1.jar:/usr/local/apache-storm-1.0.1/lib/minlog-1.3.0.jar:/usr/local/apache-storm-1.0.1/lib/reflectasm-1.10.1.jar:/usr/local/apache-storm-1.0.1/lib/clojure-1.7.0.jar:/usr/local/apache-storm-1.0.1/lib/log4j-api-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-core-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-1.0.1:/usr/local/apache-storm-1.0.1/conf -Xmx768m -Dlogfile.name=ui.log -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector -Dlog4j.configurationFile=/usr/local/apache-storm-1.0.1/log4j2/cluster.xml org.apache.storm.ui.core
网络不可达（connection reset）问题
出现网络不可达的问题，经过分析，确认问题出在hostname上，只需要修改hostname相关配置即可。
vi /etc/sysconfig/network NETWORKING=yes HOSTNAME=xxx
整体部署完成后，storm集群的拓扑结构如图：
nimbus对应多个supervisor，每个supervisor中有着固定的槽，对应不同的worker。而一个运行中的topology，由工作进程，执行器和任务组成：
在 Worker 中运行的是拓扑的一个子集。一个 worker 进程是从属于某一个特定的拓扑的，在 worker 进程中会运行一个或者多个与拓扑中的组件相关联的 executor。一个运行中的拓扑就是由这些运行于 Storm 集群中的很多机器上的进程组成的。
一个 executor 是由 worker 进程生成的一个线程。在 executor 中可能会有一个或者多个 task，这些 task 都是为同一个组件（spout 或者 bolt）服务的。
task 是实际执行数据处理的最小工作单元（注意，task 并不是线程） —— 在你的代码中实现的每个 spout 或者 bolt 都会在集群中运行很多个 task。在拓扑的整个生命周期中每个组件的 task 数量都是保持不变的，不过每个组件的 executor 数量却是有可能会随着时间变化。在默认情况下 task 的数量是和 executor 的数量一样的，也就是说，默认情况下 Storm 会在每个线程上运行一个 task。
- worker数量是拓扑topology在集群中运行所需要的工作进程数，通过Config.setNumWorker(int number)来进行设置，同一个worker从属于单个进程；
- 每个组件需要执行的线程数为executors数量，注意，不同于hadoop是以进程方式启动Map/Reducer，storm中的executor是以线程方式启动的，我们在构建Topology时，可以通过TopologyBuilder.setSpout/setBolt的数量进行控制；
- 每个组件需要的执行任务数，该参数为tasks数量，相同组件的任务执行逻辑必定是相同的。
举个例子：
topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
.setNumTasks(4)
.shuffleGrouping("blue-spout);
.setNumTasks(4)
.shuffleGrouping("blue-spout);
在上面的代码中，我们为
GreenBolt 配置了 2 个初始执行线程（executor）和 4 个关联任务（task）。这样，每个执行线程中会运行 2 个任务。如果你在设置 bolt 的时候不指定 task 的数量，那么每个 executor 的 task 数会默认设置为 1。提交storm任务用于测试
为了测试storm集群是否可以，可以提交一个测试任务，提交storm任务的命令如下（与hadoop提交任务非常类似）：
storm jar 对应的jar包 执行主类 后续参数 storm jar Getting-Started-0.0.1-SNAPSHOT.jar example.FirstTopo FirstTopo
提交完成后，在控制台中会显示：
Running: /usr/java/jdk1.7.0_60/bin/java -client -Ddaemon.name= -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-1.0.1 -Dstorm.log.dir=/usr/local/apache-storm-1.0.1/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-1.0.1/lib/servlet-api-2.5.jar:/usr/local/apache-storm-1.0.1/lib/kryo-3.0.3.jar:/usr/local/apache-storm-1.0.1/lib/asm-5.0.3.jar:/usr/local/apache-storm-1.0.1/lib/objenesis-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-rename-hack-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-1.0.1/lib/disruptor-3.3.2.jar:/usr/local/apache-storm-1.0.1/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-1.0.1/lib/log4j-core-2.1.jar:/usr/local/apache-storm-1.0.1/lib/minlog-1.3.0.jar:/usr/local/apache-storm-1.0.1/lib/reflectasm-1.10.1.jar:/usr/local/apache-storm-1.0.1/lib/clojure-1.7.0.jar:/usr/local/apache-storm-1.0.1/lib/log4j-api-2.1.jar:/usr/local/apache-storm-1.0.1/lib/storm-core-1.0.1.jar:/usr/local/apache-storm-1.0.1/lib/log4j-slf4j-impl-2.1.jar:Getting-Started-0.0.1-SNAPSHOT.jar:/usr/local/apache-storm-1.0.1/conf:/usr/local/apache-storm-1.0.1/bin -Dstorm.jar=Getting-Started-0.0.1-SNAPSHOT.jar example.FirstTopo FirstTopo
1163 [main] INFO o.a.s.StormSubmitter - Generated ZooKeeper secret payload for MD5-digest: -7925588109049404291:-4936131398353686676
1284 [main] INFO o.a.s.s.a.AuthUtils - Got AutoCreds []
1404 [main] INFO o.a.s.StormSubmitter - Uploading topology jar Getting-Started-0.0.1-SNAPSHOT.jar to assigned location: /usr/local/apache-storm-1.0.1/storm-local/nimbus/inbox/stormjar-aeffc559-6dcc-4f58-81bb-cd214fbc6512.jar
1441 [main] INFO o.a.s.StormSubmitter - Successfully uploaded topology jar to assigned location: /usr/local/apache-storm-1.0.1/storm-local/nimbus/inbox/stormjar-aeffc559-6dcc-4f58-81bb-cd214fbc6512.jar
1441 [main] INFO o.a.s.StormSubmitter - Submitting topology FirstTopo in distributed mode with conf {"storm.zookeeper.topology.auth.scheme":"digest","storm.zookeeper.topology.auth.payload":"-7925588109049404291:-4936131398353686676","topology.workers":3}
4562 [main] INFO o.a.s.StormSubmitter - Finished submitting topology: FirstTopo
同时会在storm ui中显示topology的概述信息：
Storm中的日志
Storm默认与log4j集成，可以在$STORM_HOME/log4j2目录下查找log4j的配置文件，通过修改log4j的配置文件来调整worker以及cluster的日志行为。
在Storm中所有日志都存在 logs 目录下，在nimbus节点上，nimbus.log记录的是nimbus启动过程中的输出信息，包括启动时间以及各个worker和task初始化过程中打印信息等等，ui.log记录是storm监控程序启动过程中的输出信息，包括启动时间；在supervisor节点上，supervisor.log记录的是supervisor启动的相关信息，以下面的目录举例说明一下，/usr/local/apache-storm-1.0.1是STORM_HOME，存在workers-artifacts目录下，MyKafkaTopology为storm任务名称，MyKafkaTopology-4-1467270486为该任务在storm中的任务id，6703为该任务的端口
/usr/local/apache-storm-1.0.1/logs/workers-artifacts/MyKafkaTopology-4-1467270486/6703
通过启动supervisor上的logviewer(执行命令 storm logviewer）可以以图形化方式查看supervisor上运行的日志，默认启动的http端口为8000。
下面就是其中的一个Task的所有executor，通过点击端口号，可以查看storm的实时日志：
其中包含多种类型日志：
gc.log.0 worker.log worker.log.err worker.log.metrics worker.log.out worker.pid worker.yaml
在程序中加入日志，可以通过log4j的方式：
import org.apache.log4j.Logger;
private static final Logger LOG = Logger.getLogger(KafkaWordSplitter.class);
但是将jar包提交到storm集群上，在执行过程中出现日志使用错误：
Exception in thread "main" java.lang.ExceptionInInitializerError
at org.apache.log4j.Logger.getLogger(Logger.java:39)
at org.apache.log4j.Logger.getLogger(Logger.java:43)
at com.zhen.storm.example.MyKafkaTopology$KafkaWordSplitter.<clinit>(MyKafkaTopology.java:33)
at com.zhen.storm.example.MyKafkaTopology.main(MyKafkaTopology.java:118)
Caused by: java.lang.IllegalStateException: Detected both log4j-over-slf4j.jar AND slf4j-log4j12.jar on the class path, preempting *Error. See also http://www.slf4j.org/codes.html#log4jDelegationLoop for more details.
at org.apache.log4j.Log4jLoggerFactory.<clinit>(Log4jLoggerFactory.java:49)
... 4 more
参考文档：http://blog.csdn.net/gongmf/article/details/40379547 来解决此问题，maven依赖的kafka包存在slf4j-log4j12的相关依赖，将其exclude出去即可：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.10.0.0</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
关于storm的配置项详解可以参考：http://xstarcd.github.io/wiki/Cloud/storm_config_detail.html
