| 编辑推荐: |
| 本文来自于jdon，本文主要介绍了Storm、Storm和Hadoop的区别、Storm的容错性和可靠性等相关内容。 |
|
什么是Storm？
Storm是：
快速且可扩展伸缩
容错
确保消息能够被处理
易于设置和操作
开源的分布式实时计算系统
- 最初由Nathan Marz开发
- 使用Java 和 Clojure 编写
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();
}
} |
在这个代码中定义了一些参数比如Works的数目是4，其含义在后面详细分析。
下面我们要将上面这段代码发布部署到Storm中，首先了解Storm物理架构图：
Nimbus是一个主后台处理器，主要负责：
1.发布分发代码
2.分配任务
3.监控失败。
Supervisor是负责当前这个节点的后台工作处理器的监听。
Work类似Java的线程，采取JDK的Executor 。
下面开始将我们的代码部署到这个网络拓扑中：
将代码Jar包上传到Nimbus的inbox，包括所有的依赖包，然后提交。
Nimbus将保存在本地文件系统，然后开始配置网络拓扑，分配开始拓扑。
见下图：
Nimbus服务器将拓扑Jar 配置和结构下载到 Supervisor，负载平衡ZooKeeper分配某个特定的Supervisor服务器，而Supervisor开始基于配置分配Work，Work调用JDK的Executor启动线程，开始任务处理。
下面是我们代码对拓扑分配的参数示意图：
Executor启动的线程数目是12个，组件的实例是16个，那么如何在实际服务器中分配呢？如下图：
图中RsSpout代表我们的代码中RandomSentenceSpout；SplitSentenceBolt简写为SSbolt；
现在开始分析Storm内部架构，首先看看Work之间的消息传递，如下图，
Work之间的通讯是通过ZeroMQ，但是Yahoo后来发现使用异步的Netty能够提升Storm一倍性能，数据使用Kryo进行序列化，本地通讯使用Lmax的Disruptor
，内部无需序列化。
容错性
如下图，executor发送心跳到Zookeeper，Supervisor从本地文件读取所在服务器的worker心跳状态，然后同步分配发送到zooKeeper。Nimbus监控集群状态。这样能确保worker一直活着。
如果某个节点也就是服务器没有心跳，Nimbus将重新分配新的服务器上线工作。
如果某个节点服务器中work没有心跳，那么Supervisor将负责重启线程。
如果某个Supervisor完蛋，整个处理正常，但是分配的同步工作就无法进行了。
如果Nimbus崩溃，整个系统可以运行，但是拓扑分配工作无法进行了。
可靠性：确保消息被处理
public class
RandomSentenceSpout extends BaseRichSpout {
public void nextTuple() {
UUID msgId = getMsgId();//用消息ID发送消息
collector.emit(new Values(tweet), msgId);
}
public void ack(Object msgId) {
// Do something with acked message id. 确认消息ID
}
public void fail(Object msgId) {
// Do something with failed message id. 消息ID失败了
}
}
public class SplitSentenceBolt extends BaseRichBolt
{
public void execute(Tuple input) {
for (String s : input.getString(0).split("\\s"))
{
collector.emit(input, new Values(s));
}
//当整个词语都被切分后，确认输入的事件已经被接受处理。
collector.ack(input);
}
} |
下面是一个Ack确认流程，注意到Acker Implicit bolt。
对于一个树形结构Tuple流，也就是Tuple里面嵌套Tuple。
如果事件被下一个节点成功接受和处理，这个节点将更新相应初始事件的签名，通过异或操作，将输入事件的ID和所有基于该输入事件产生的所有事件的ID进行异或操作，如下图，事件
01111 产生子事件 01100, 10010, 和 00010, 这样事件 01111的签名是11100
(= 01111 (initial value) xor 01111 xor 01100 xor 10010
xor 00010).
当Ack值变成0，Acker implicit bolt就知道tuple树形数据集合全部被处理完成，一个事务确保可靠结束。例如语句分成一个个单词全部完成。
Storm的集群设置
设置ZooKeeper cluster
(1)安装Storm依赖的库包到服务器上:
- ZeroMQ 2.1.7 and JZMQ
- Java 6 and Python 2.6.6
- unzip
(2)下载解压Storm。
填写强制性配置到storm.yaml
用storm脚本启动守护流程的监督
通过Web界面能够观察管理拓扑网络情况和组件情况。 |
