让你真正明白什么是storm

about云腾讯认证空间

本帖最后由 bioger_hit 于 2014-10-17 11:30 编辑

问题导读：
1.你认为什么图形可以显示hadoop与storm的区别？
2.本文是如何形象讲解hadoop与storm的？
3.hadoop map/reduce对应storm那两个概念？
4.storm流由谁来组成？
5.tuple具体是什么形式？

个人总结
当我们在学习一项新的技术时，由于大量信息的涌入，导致我们什么也记不住，如果初学storm，并且对于hadoop有所了解，那么只要知道hadoop中map/reduce所对应storm中的Spout/bolt即可，
那么明白了这两个概念有什么用？作用还是比较大的，因为最起码已经算是storm编程入门了。我们会编写maprecue，那么hadoop基本编程没有问题，所以如果我们会编写Spout/bolt，那么我们就会storm编程了，这是多么简单的一件事情。

更多的概念可以慢慢消化，比如hadoop有master和slave，那么strom是如何对应的，这些疑问可以留作以后。

本文以电梯的方式讲解了hadoop与storm的区别，凸显了它们一个为批量处理，一个为流式处理。

更详细内容参考下面：

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

我们知道hadoop是批处理，storm是流式处理，那么是什么是批处理，什么刘处理？
Storm和Hadoop主要区别是实时和批处理的区别：

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

Tuple是Storm的数据模型，如['jdon',12346]
多个Tuple组成事件流：

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

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

这些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();
}
}
复制代码