41.storm搭建

伪分布式

先检查配置环境，Python版本
Java 6+
Python 2.6.6+

上传、解压安装包

修改配置文件

添加环境变量
storm伪分布式已经搭好.png

新建一个文件mkdir logs，是为了把信息输出到logs文件里

storm - dev-zookeeper >> （输出到）./logs/zk.out 2（标准的错误输出）>&1（定向到标准的输出） &(后台运行)

jps查看进程
启动会耗费一些时间

同样启动nimbus，supervisor和ui

都启动完毕了

要查看端口，8080在运行才表示真正运行了

web界面

打jar包上传到storm里，运行
15.png
运行结果

全分布式

全分布式和伪分布式最大的区别就是配置问题

修改storm的配置

修改zk，存放路径（storm.local.dir）

分发

开启zk

上面几张图片就类似于伪分布式

——————————————————————————————————————————————
storm部署文档
一、环境要求
JDK 1.6+
java -version
Python 2.6.6+
python -V

ZooKeeper3.4.5+
storm 0.9.4+

---

二、单机模式
上传解压
$ tar xf apache-storm-0.9.4.tar.gz
$ cd apache-storm-0.9.4

$ storm安装目录下创建log： mkdir logs
$ ./bin/storm --help
下面分别启动ZooKeeper、Nimbus、UI、supervisor、logviewer
$ ./bin/storm dev-zookeeper >> ./logs/zk.out 2>&1 &
$ ./bin/storm nimbus >> ./logs/nimbus.out 2>&1 &
$ ./bin/storm ui >> ./logs/ui.out 2>&1 &
$ ./bin/storm supervisor >> ./logs/supervisor.out 2>&1 &
$ ./bin/storm logviewer >> ./logs/logviewer.out 2>&1 &
需要等一会儿
$ jps
6966 Jps
6684 logviewer
6680 dev_zookeeper
6681 nimbus
6682 core
6683 supervisor

http://node01:8080
提交任务到Storm集群当中运行：
$ ./bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.4.jar storm.starter.WordCountTopology wordcount
$ ./bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.4.jar storm.starter.WordCountTopology test

---

三、完全分布式安装部署
各节点分配：
Nimbus Supervisor Zookeeper
node1 ( Nimbus,Zookeeper) 1 1
node2 ( Supervisor,Zookeeper) 1 1
node3 ( Supervisor,Zookeeper) 1 1

node1作为nimbus，
开始配置
$ vim conf/storm.yaml
storm.zookeeper.servers:
- "node1"
- "node2"
- "node3"

storm.local.dir: "/tmp/storm"

nimbus.host: "node1"

supervisor.slots.ports:
- 6700
- 6701
- 6702
- 6703

在storm目录中创建logs目录
$ mkdir logs

（分发）集群其他服务器

启动ZooKeeper集群

node1上启动Nimbus
$ ./bin/storm nimbus >> ./logs/nimbus.out 2>&1 &
$ tail -f logs/nimbus.log
$ ./bin/storm ui >> ./logs/ui.out 2>&1 &
$ tail -f logs/ui.log

节点node2和node3启动supervisor，按照配置，每启动一个supervisor就有了4个slots
$ ./bin/storm supervisor >> ./logs/supervisor.out 2>&1 &
$ tail -f logs/supervisor.log
（当然node1也可以启动supervisor）

http://node1:8080/
提交任务到Storm集群当中运行：
$ ./bin/storm jar examples/storm-starter/storm-starter-topologies-0.9.4.jar storm.starter.WordCountTopology test

环境变量可以配置也可以不配置
export STORM_HOME=/opt/sxt/storm
export PATH=$PATH:$STORM_HOME/bin

观察关闭一个supervisor后，nimbus的重新调度
再次启动一个新的supervisor后，观察，并rebalance

集群drpc

修改
$ vi conf/storm.yaml
drpc.servers:
- "node06"

分发配置storm.yaml文件给其他节点

启动zk
主节点启动 nimbus,supervisor,drpc
从启动 supervisor

---

sotrm并发机制
Worker – 进程

• 一个Topology拓扑会包含一个或多个Worker（每个Worker进程只能从属于一个特定的Topology）
• 这些Worker进程会并行跑在集群中不同的服务器上，即一个Topology拓扑其实是由并行运行在Storm集群中多台服务器上的进程所组成

Executor – 线程

• Executor是由Worker进程中生成的一个线程
• 每个Worker进程中会运行拓扑当中的一个或多个Executor线程
• 一个Executor线程中可以执行一个或多个Task任务（默认每个Executor只执行一个Task任务），但是这些Task任务都是对应着同一个组件（Spout、Bolt）。

Task

• 实际执行数据处理的最小单元
• 每个task即为一个Spout或者一个Bolt

Task数量在整个Topology生命周期中保持不变，Executor数量可以变化或手动调整
（默认情况下，Task数量和Executor是相同的，即每个Executor线程中默认运行一个Task任务）

设置Worker进程数

• Config.setNumWorkers(int workers)

设置Executor线程数

• TopologyBuilder.setSpout(String id, IRichSpout spout, Number parallelism_hint)
• TopologyBuilder.setBolt(String id, IRichBolt bolt, Number parallelism_hint)
• ：其中， parallelism_hint即为executor线程数

设置Task数量

• ComponentConfigurationDeclarer.setNumTasks(Number val)

例：
Config conf = new Config() ;
conf.setNumWorkers(2);

TopologyBuilder topologyBuilder = new TopologyBuilder();
topologyBuilder.setSpout("spout", new MySpout(), 1);
topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
.setNumTasks(4)
.shuffleGrouping("blue-spout);

图片1.png 图片2.png