Storm - 大数据Big Data实时处理架构，在Windows上安装Storm

偏执的太偏执、 2022-04-21 14:52 255阅读 0赞

# Storm - 大数据Big Data实时处理架构 #

### 什么是Storm？ ###

Storm是：  
• 快速且可扩展伸缩  
• 容错  
• 确保消息能够被处理  
• 易于设置和操作  
• 开源的分布式实时计算系统  
\- 最初由Nathan Marz开发  
\- 使用Java 和 Clojure 编写

Storm和Hadoop主要区别是实时和批处理的区别：

![storm hadoop][]

Storm概念 组成：Spout 和Bolt组成Topology。

![storm][]

Tuple是Storm的数据模型，如\['jdon',12346\]

多个Tuple组成事件流：

![tuple][]

Spout是读取需要分析处理的数据源，然后转为Tuples，这些数据源可以是Web日志、 API调用、数据库等等。Spout相当于事件流的生产者。

Bolt 处理Tuples然后再创建新的Tuples流，Bolt相当于事件流的消费者。

Bolt 作为真正业务处理者，主要实现大数据处理的核心功能，比如转换数据，应用相应过滤器，计算和聚合数据(比如统计总和等等) 。

以Twitter的某个Tweet为案例，看看Storm如何处理：

![top][]

这些tweett贴内容是：“No Small Cell Lung \#Cancer(没有小细胞肺癌＃癌症)” "An \#OnCology Consult...."

这些贴被Spout读取以后，产生Tuple，字段名是tweet，内容是"No Small Cell Lung \#Cancer"，格式类似：\['No Small Cell Lung \#Cancer',133221\]。

然后进入被流 消费者Bolt进行处理，第一个Bolt是SplitSentence，将tuple内容进行分离，结果成为：一个个单词："No" "Small" "Cell" "Lung" "\#Cancer" ；然后经过第二个Bolt进行过滤HashTagFilter处理，Hash标签是单词中用\#标注的，也就是Cancer；再经过HasTagCount计数，可以本地内存缓存这个计数结果，最后通过PrinterBolt打印出标签单词统计结果 。

我们使用Stom所要做的就是编制Spout和Bolt代码：

public class RandomSentenceSpout extends BaseRichSpout \{  
　　SpoutOutputCollector collector;  
　　Random random;  
  
　　//读入外部数据  
　　public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) \{  
　　　　this.collector = collector;  
　　　　random = new Random();  
　　\}  
　　//产生Tuple  
　　 public void nextTuple() \{  
　　　　String\[\] sentences = new String\[\] \{  
　　　　　　"No Small Cell Lung \#Cancer",  
　　　　　　"An \#OnCology Consultant 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"  
　　　　\};  
　　　　String tweet = sentences\[random.nextInt(sentences.length)\];  
　　　　//定义字段名"tweet" 的值  
　　　　collector.emit(new Values(tweet));

// 定义字段名"tweet"

public void declareOutputFields(OutputFieldsDeclarer declarer) \{  
　　　　declarer.declare(new Fields("tweet"));  
　　\}  
　　@Override  
　　public void ack(Object msgId) \{\}  
　　@Override  
　　public void fail(Object msgId) \{\}  
\}

下面是Bolt的代码编写：

public class SplitSentenceBolt extends BaseRichBolt \{  
　　OutputCollector collector;  
  
　　@Override  
　　public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) \{  
　　　　this.collector = collector;  
　　\}  
　　@Override 消费者激活主要方法：分离成单个单词  
　　public void execute(Tuple input) \{  
　　　　for (String s : input.getString(0).split("\\\\s")) \{  
　　　　　　collector.emit(new Values(s));  
　　　　\}  
　　\}  
　　@Override 定义新的字段名  
　　public void declareOutputFields(OutputFieldsDeclarer declarer) \{  
　　　　declarer.declare(new Fields("word"));  
　　\}

最后是装配运行Spout和Bolt的客户端调用代码：

public class WordCountTopology \{  
　　public static void main(String\[\] args) throws Exception \{  
　　　　TopologyBuilder builder = new TopologyBuilder();  
　　　　builder.setSpout("tweet", new RandomSentenceSpout(), 2);  
　　　　builder.setBolt("split", new SplitSentenceBolt(), 4)  
　　　　　　.shuffleGrouping("tweet")  
　　　　　　.setNumTasks(8);  
　　　　builder.setBolt("count", new WordCountBolt(), 6)  
　　　　　　.fieldsGrouping("split", new Fields("word"));  
　　　　..设置多个Bolt

Config config = new Config();  
　　　　config.setNumWorkers(4);  
　　　　  
　　　　StormSubmitter.submitTopology("wordcount", config, builder.createTopology());

// Local testing  
//LocalCluster cluster = new LocalCluster();  
// cluster.submitTopology("wordcount", config, builder.createTopology());  
//Thread.sleep(10000);  
//cluster.shutdown();  
\}  
\}

在这个代码中定义了一些参数比如Works的数目是4，其含义在后面详细分析。

下面我们要将上面这段代码发布部署到Storm中，首先了解Storm物理架构图：

![stormph.png][]

Nimbus是一个主后台处理器，主要负责：  
1.发布分发代码  
2.分配任务  
3.监控失败。

Supervisor是负责当前这个节点的后台工作处理器的监听。

Work类似Java的线程，采取JDK的Executor 。

下面开始将我们的代码部署到这个网络拓扑中：

![startstorm][]

将代码Jar包上传到Nimbus的inbox，包括所有的依赖包，然后提交。

Nimbus将保存在本地文件系统，然后开始配置网络拓扑，分配开始拓扑。

见下图：

![storm][storm 1]

Nimbus服务器将拓扑Jar 配置和结构下载到 Supervisor，负载平衡ZooKeeper分配某个特定的Supervisor服务器，而Supervisor开始基于配置分配Work，Work调用JDK的Executor启动线程，开始任务处理。

下面是我们代码对拓扑分配的参数示意图：

![storm bolt][]

Executor启动的线程数目是12个，组件的实例是16个，那么如何在实际服务器中分配呢？如下图：

![wc][]

图中RsSpout代表我们的代码中RandomSentenceSpout；SplitSentenceBolt简写为SSbolt；

# 在Windows上安装Storm #

首先安装Java。下载 [JDK 7 from Oracle][].，比如安装在

C:\Java\jdk1.7.0_45\

### 安装 Python ###

这是为了测试安装效果，我们将部署 [storm-starter project][]案例中word coun程序，用的是python写的[multi-lang][] bolt，使用python 2.7.6，安装路径在：

C:\Python27\

### 安装运行ZooKeeper ###

[Download Apache Zookeeper 3.3.6][] ，解压配置：

> cd zookeeper-3.3.6  
    > copy conf\zoo_sample.cfg conf\zoo.cfg  
    > .\bin\zkServer.cmd

###    安装Storm ###

Storm的windows官方版还没有释放，[here][].下载，源码[here][here 1]下载

### 配置环境 ###

Storm需要STORM\_HOME和JAVA\_HOME，比如

STORM\_HOME为

C:\storm-0.9.1-incubating-SNAPSHOT-12182013\

在PATH中加入：

%STORM\_HOME%\\bin;%JAVA\_HOME%\\bin;C:\\Python27;C:\\Python27\\Lib\\site-packages\\;C:\\Python27\\Scripts\\;

PATHEXT加入

.PY

### 启动Nimbus, Supervisor, and Storm UI Daemons ###

**Nimbus**

> cd %STORM_HOME%  > storm nimbus

**Supervisor**

> cd %STORM_HOME%  > storm supervisor

**Storm UI**

> cd %STORM_HOME%  > storm ui

浏览器打开[http://localhost:8080/][http_localhost_8080] 可看到Storm运行。

### 部署Word count ###

下载[download a pre-built jar][]。

部署这个jar在本地：

> storm jar storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar storm.starter.WordCountTopology WordCount -c nimbus.host=localhost

如果你刷新 Storm UI页面，会看到 “WordCount” topology显示列出，点按链接确认它处理数据。

[storm hadoop]: /images/20220414/14bcd1ac88ca4bcb82b18ba6fd31e403.png
[storm]: https://www.jdon.com/simgs/bigdata/storm.png
[tuple]: https://www.jdon.com/simgs/bigdata/tuples.png
[top]: https://www.jdon.com/simgs/top.png
[stormph.png]: https://www.jdon.com/simgs/bigdata/stormph.png
[startstorm]: https://www.jdon.com/simgs/bigdata/startStorm.png
[storm 1]: https://www.jdon.com/simgs/bigdata/sstorm.png
[storm bolt]: https://www.jdon.com/simgs/bigdata/stormbolt.png
[wc]: https://www.jdon.com/simgs/bigdata/wc.png
[JDK 7 from Oracle]: http://download.oracle.com/otn-pub/java/jdk/7u45-b18/jdk-7u45-windows-x64.exe
[storm-starter project]: https://github.com/nathanmarz/storm-starter
[multi-lang]: https://github.com/nathanmarz/storm/wiki/Multilang-protocol
[Download Apache Zookeeper 3.3.6]: http://download.nextag.com/apache/zookeeper/zookeeper-3.3.6/
[here]: https://dl.dropboxusercontent.com/s/iglqz73chkul1tu/storm-0.9.1-incubating-SNAPSHOT-12182013.zip
[here 1]: https://github.com/ptgoetz/incubator-storm/tree/windows-test
[http_localhost_8080]: http://localhost:8080/
[download a pre-built jar]: https://dl.dropboxusercontent.com/s/kc933u6vz2crqkb/storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar