Storm实时计算：流操作入门编程实践
- - 简单之美Storm是一个分布式是实时计算系统，它设计了一种对流和计算的抽象，概念比较简单，实际编程开发起来相对容易. 下面，简单介绍编程实践过程中需要理解的Storm中的几个概念：. 一个Topology运行以后就不能停止，它会无限地运行下去，除非手动干预（显式执行bin/storm kill ）或意外故障（如停机、整个Storm集群挂掉）让它终止.
Storm实时计算：流操作入门编程实践
标签：
Storm
开源技术
| 发表时间：2014-11-28 13:16 | 作者：Yanjun
出处：http://shiyanjun.cn
Storm是一个分布式是实时计算系统，它设计了一种对流和计算的抽象，概念比较简单，实际编程开发起来相对容易。下面，简单介绍编程实践过程中需要理解的Storm中的几个概念：
- Topology
Storm中Topology的概念类似于Hadoop中的MapReduce Job，是一个用来编排、容纳一组计算逻辑组件（Spout、Bolt）的对象（Hadoop MapReduce中一个Job包含一组Map Task、Reduce Task），这一组计算组件可以按照DAG图的方式编排起来（通过选择Stream Groupings来控制数据流分发流向），从而组合成一个计算逻辑更加负责的对象，那就是Topology。一个Topology运行以后就不能停止，它会无限地运行下去，除非手动干预（显式执行bin/storm kill ）或意外故障（如停机、整个Storm集群挂掉）让它终止。
- Spout
Storm中Spout是一个Topology的消息生产的源头，Spout应该是一个持续不断生产消息的组件，例如，它可以是一个Socket Server在监听外部Client连接并发送消息，可以是一个消息队列（MQ）的消费者、可以是用来接收Flume Agent的Sink所发送消息的服务，等等。Spout生产的消息在Storm中被抽象为Tuple，在整个Topology的多个计算组件之间都是根据需要抽象构建的Tuple消息来进行连接，从而形成流。
- Bolt
Storm中消息的处理逻辑被封装到Bolt组件中，任何处理逻辑都可以在Bolt里面执行，处理过程和普通计算应用程序没什么区别，只是需要根据Storm的计算语义来合理设置一下组件之间消息流的声明、分发、连接即可。Bolt可以接收来自一个或多个Spout的Tuple消息，也可以来自多个其它Bolt的Tuple消息，也可能是Spout和其它Bolt组合发送的Tuple消息。
- Stream Grouping
Storm中用来定义各个计算组件（Spout、Bolt）之间流的连接、分组、分发关系。Storm定义了如下7种分发策略：Shuffle Grouping（随机分组）、Fields Grouping（按字段分组）、All Grouping（广播分组）、Global Grouping（全局分组）、Non Grouping（不分组）、Direct Grouping（直接分组）、Local or Shuffle Grouping（本地/随机分组），各种策略的具体含义可以参考Storm官方文档、比较容易理解。
下面，作为入门实践，我们简单介绍几种开发中常用的流操作处理方式的实现：
Storm组件简单串行
这种方式是最简单最直观的，只要我们将Storm的组件（Spout、Bolt）串行起来即可实现，只需要了解编写这些组件的基本方法即可。在实际应用中，如果我们需要从某一个数据源连续地接收消息，然后顺序地处理每一个请求，就可以使用这种串行方式来处理。如果说处理单元的逻辑非常复杂，那么就需要处理逻辑进行分离，属于同一类操作的逻辑封装到一个处理组件中，做到各个组件之间弱耦合（除了定义Field的schema外，只通过发送消息来连接各个组件）。
下面，我实现一个简单的WordCount的例子，各个组件之间的连接方式，如下图所示：
ProduceRecordSpout类是一个Spout组件，用来产生消息，我们这里模拟发送一些英文句子，实际应用中可以指定任何数据源，如数据库、消息中间件、Socket连接、RPC调用等等。ProduceRecordSpout类代码如下所示：
public static class ProduceRecordSpout extends BaseRichSpout {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(ProduceRecordSpout.class);
private SpoutOutputCollector collector;
private Random random;
private String[] records;
public ProduceRecordSpout(String[] records) {
this.records = records;
}
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
random = new Random();
}
@Override
public void nextTuple() {
Utils.sleep(500);
String record = records[random.nextInt(records.length)];
List<Object> values = new Values(record);
collector.emit(values, values);
LOG.info("Record emitted: record=" + record);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("record"));
}
}
构造一个ProduceRecordSpout对象时，传入一个字符串数组，然后随机地选择其中一个句子，emit到下游（Downstream）的WordSplitterBolt组件，只声明了一个Field，WordSplitterBolt组件可以根据声明的Field，接收到emit的消息，WordSplitterBolt类代码实现如下所示：
public static class WordSplitterBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(WordSplitterBolt.class);
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String record = input.getString(0);
if(record != null && !record.trim().isEmpty()) {
for(String word : record.split("\\s+")) {
collector.emit(input, new Values(word, 1));
LOG.info("Emitted: word=" + word);
collector.ack(input);
}
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word", "count"));
}
}
在execute方法中，传入的参数是一个Tuple，该Tuple就包含了上游（Upstream）组件ProduceRecordSpout所emit的数据，直接取出数据进行处理。上面代码中，我们将取出的数据，按照空格进行的split，得到一个一个的单词，然后在emit到下一个组件，声明的输出schema为2个Field：word和count，当然这里面count的值都为1。
进行统计词频的组件为WordCounterBolt，实现代码如下所示：
public static class WordCounterBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(WordCounterBolt.class);
private OutputCollector collector;
private final Map<String, AtomicInteger> counterMap = Maps.newHashMap();
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String word = input.getString(0);
int count = input.getIntegerByField("count"); // 通过Field名称取出对应字段的数据
AtomicInteger ai = counterMap.get(word);
if(ai == null) {
ai = new AtomicInteger(0);
counterMap.put(word, ai);
}
ai.addAndGet(count);
LOG.info("DEBUG: word=" + word + ", count=" + ai.get());
collector.ack(input);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
@Override
public void cleanup() {
// print count results
LOG.info("Word count results:");
for(Entry<String, AtomicInteger> entry : counterMap.entrySet()) {
LOG.info("\tword=" + entry.getKey() + ", count=" + entry.getValue().get());
}
}
}
上面代码通过一个Map来对每个单词出现的频率进行累加计数，比较简单。因为该组件是Topology的最后一个组件，所以不需要在declareOutputFields方法中声明Field的Schema，而是在cleanup方法中输出最终的结果，只有在该组件结束任务退出时才会调用cleanup方法输出。
最后，需要基于上面的3个组件来创建一个Topology实例，提交到Storm集群去运行，配置代码如下所示：
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
// configure & build topology
TopologyBuilder builder = new TopologyBuilder();
String[] records = new String[] {
"A Storm cluster is superficially similar to a Hadoop cluster",
"All coordination between Nimbus and the Supervisors is done through a Zookeeper cluster",
"The core abstraction in Storm is the stream"
};
builder
.setSpout("spout-producer", new ProduceRecordSpout(records), 1)
.setNumTasks(3);
builder
.setBolt("bolt-splitter", new WordSplitterBolt(), 2)
.shuffleGrouping("spout-producer")
.setNumTasks(2);
builder.setBolt("bolt-counter", new WordCounterBolt(), 1)
.fieldsGrouping("bolt-splitter", new Fields("word"))
.setNumTasks(2);
// submit topology
Config conf = new Config();
String name = WordCountTopology.class.getSimpleName();
if (args != null && args.length > 0) {
String nimbus = args[0];
conf.put(Config.NIMBUS_HOST, nimbus);
conf.setNumWorkers(2);
StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(name, conf, builder.createTopology());
Thread.sleep(60000);
cluster.shutdown();
}
}
上面通过TopologyBuilder来配置组成一个Topology的多个组件（Spout或Bolt），然后通过调用createTopology()方法创建一个Topology实例。上面方法中，对应着2种运行模式：如果没有传递任何参数，则是使用LocalCluster来运行，适合本地调试代码；如果传递一个Topology名称作为参数，则是在真实的Storm集群上运行，需要对实现的Topology代码进行编译打包，通过StormSubmitter提交到集群上作为服务运行。
Storm组合多种流操作
Storm支持流聚合操作，将多个组件emit的数据，汇聚到同一个处理组件来统一处理，可以实现对多个Spout组件通过流聚合到一个Bolt组件（Sout到Bolt的多对一、多对多操作），也可以实现对多个Bolt通过流聚合到另一个Bolt组件（Bolt到Bolt的多对一、多对多操作）。实际，这里面有两种主要的操作，一种是类似工作流中的fork，另一种是类似工作流中的join。下面，我们实现一个例子来演示如何使用，实时流处理逻辑如下图所示：
上图所描述的实时流处理流程，我们期望能够按照如下流程进行处理：
- 存在3类数据：数字字符串（NUM）、字母字符串（STR）、特殊符号字符串（SIG）
- 每个ProduceRecordSpout负责处理上面提到的3类数据
- 所有数据都是字符串，字符串中含有空格，3种类型的ProduceRecordSpout所emit的数据都需要被相同的逻辑处理：根据空格来拆分字符串
- 一个用来分发单词的组件DistributeWordByTypeBolt能够接收到所有的单词（包含类型信息），统一将每类单词分别分发到指定的一个用来存储数据的组件
- SaveDataBolt用来存储处理过的单词，对于不同类型单词具有不同的存储逻辑，需要设置3类SaveDataBolt
将Spout分为3类，每一个Spout发射不同类型的字符串，这里定义了一个Type常量类来区分这三种类型：
interface Type {
String NUMBER = "NUMBER";
String STRING = "STRING";
String SIGN = "SIGN";
}
首先看一下，我们实现的Topology是如何进行创建的，代码如下所示：
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException {
// configure & build topology
TopologyBuilder builder = new TopologyBuilder();
// configure 3 spouts
builder.setSpout("spout-number", new ProduceRecordSpout(Type.NUMBER, new String[] {"111 222 333", "80966 31"}), 1);
builder.setSpout("spout-string", new ProduceRecordSpout(Type.STRING, new String[] {"abc ddd fasko", "hello the word"}), 1);
builder.setSpout("spout-sign", new ProduceRecordSpout(Type.SIGN, new String[] {"++ -*% *** @@", "{+-} ^#######"}), 1);
// configure splitter bolt
builder.setBolt("bolt-splitter", new SplitRecordBolt(), 2)
.shuffleGrouping("spout-number")
.shuffleGrouping("spout-string")
.shuffleGrouping("spout-sign");
// configure distributor bolt
builder.setBolt("bolt-distributor", new DistributeWordByTypeBolt(), 6)
.fieldsGrouping("bolt-splitter", new Fields("type"));
// configure 3 saver bolts
builder.setBolt("bolt-number-saver", new SaveDataBolt(Type.NUMBER), 3)
.shuffleGrouping("bolt-distributor", "stream-number-saver");
builder.setBolt("bolt-string-saver", new SaveDataBolt(Type.STRING), 3)
.shuffleGrouping("bolt-distributor", "stream-string-saver");
builder.setBolt("bolt-sign-saver", new SaveDataBolt(Type.SIGN), 3)
.shuffleGrouping("bolt-distributor", "stream-sign-saver");
// submit topology
Config conf = new Config();
String name = MultiStreamsWordDistributionTopology.class.getSimpleName();
if (args != null && args.length > 0) {
String nimbus = args[0];
conf.put(Config.NIMBUS_HOST, nimbus);
conf.setNumWorkers(3);
StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(name, conf, builder.createTopology());
Thread.sleep(60 * 60 * 1000);
cluster.shutdown();
}
}
一个SplitRecordBolt组件从3个不同类型的ProduceRecordSpout接收数据，这是一个多Spout流聚合。SplitRecordBolt将处理后的数据发送给DistributeWordByTypeBolt组件，然后根据收到的数据的类型进行一个分发处理，这里用了fieldsGrouping操作，也就是SplitRecordBolt发送的数据会按照类型发送到不同的DistributeWordByTypeBolt任务（Task），每个Task收到的一定是同一个类型的数据，如果直接使用shuffleGrouping操作也没有问题，只不过每个Task可能收到任何类型的数据，在DistributeWordByTypeBolt内部进行流向控制。DistributeWordByTypeBolt组件中定义了多个stream，根据类型来分组发送给不同类型的SaveDataBolt组件。
下面看每个组件的实现：
- ProduceRecordSpout组件
通过我们定义的一个ProduceRecordSpout类，可以创建3个不同的ProduceRecordSpout实例，每个实例负责生产特定类型的数据，实现代码如下所示：
public static class ProduceRecordSpout extends BaseRichSpout {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(ProduceRecordSpout.class);
private SpoutOutputCollector collector;
private Random rand;
private String[] recordLines;
private String type;
public ProduceRecordSpout(String type, String[] lines) {
this.type = type;
recordLines = lines;
}
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
rand = new Random();
}
@Override
public void nextTuple() {
Utils.sleep(500);
String record = recordLines[rand.nextInt(recordLines.length)];
List<Object> values = new Values(type, record);
collector.emit(values, values);
LOG.info("Record emitted: type=" + type + ", record=" + record);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("type", "record"));
}
}
这比较简单，根据传递的参数来创建上图中的3个Spout实例。
- SplitRecordBolt组件
由于前面3个ProduceRecordSpout产生的数据，在开始时的处理逻辑是相同的，所以可以将3个ProduceRecordSpout聚合到一个包含通用逻辑的SplitRecordBolt组件，实现如下所示：
public static class SplitRecordBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(SplitRecordBolt.class);
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String type = input.getString(0);
String line = input.getString(1);
if(line != null && !line.trim().isEmpty()) {
for(String word : line.split("\\s+")) {
collector.emit(input, new Values(type, word));
LOG.info("Word emitted: type=" + type + ", word=" + word);
// ack tuple
collector.ack(input);
}
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("type", "word"));
}
}
无论接收到的Tuple是什么类型（STRING、NUMBER、SIGN）的数据，都进行split，然后在emit的时候，仍然将类型信息传递给下一个Bolt组件。
- DistributeWordByTypeBolt组件
DistributeWordByTypeBolt组件只是用来分发Tuple，通过定义Stream，将接收到的Tuple发送到指定的下游Bolt组件进行处理。通过SplitRecordBolt组件emit的Tuple包含了类型信息，所以在DistributeWordByTypeBolt中根据类型来进行分发，代码实现如下：
public static class DistributeWordByTypeBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(DistributeWordByTypeBolt.class);
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
Map<GlobalStreamId, Grouping> sources = context.getThisSources();
LOG.info("sources==> " + sources);
}
@Override
public void execute(Tuple input) {
String type = input.getString(0);
String word = input.getString(1);
switch(type) {
case Type.NUMBER:
emit("stream-number-saver", type, input, word);
break;
case Type.STRING:
emit("stream-string-saver", type, input, word);
break;
case Type.SIGN:
emit("stream-sign-saver", type, input, word);
break;
default:
// if unknown type, record is discarded.
// as needed, you can define a bolt to subscribe the stream 'stream-discarder'.
emit("stream-discarder", type, input, word);
}
// ack tuple
collector.ack(input);
}
private void emit(String streamId, String type, Tuple input, String word) {
collector.emit(streamId, input, new Values(type, word));
LOG.info("Distribution, typed word emitted: type=" + type + ", word=" + word);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declareStream("stream-number-saver", new Fields("type", "word"));
declarer.declareStream("stream-string-saver", new Fields("type", "word"));
declarer.declareStream("stream-sign-saver", new Fields("type", "word"));
declarer.declareStream("stream-discarder", new Fields("type", "word"));
}
}
实际上，下游的3个Bolt组件（SaveDataBolt）在订阅该流组件（DistributeWordByTypeBolt）的时候，方式相同，只是分发的逻辑交由DistributeWordByTypeBolt来统一控制。
我们在配置该Bolt组件时，使用了fieldsGrouping分组方式，实际每个DistributeWordByTypeBolt只会收到同一种类型的Tuple，这里也可以使用shuffleGrouping分组方式，这种分组方式会有不同类型的Tuple被emit到同一个DistributeWordByTypeBolt组件上。
另外，该Bolt组件中我们还定义了一个名称为stream-discarder的stream，在Topology中并没有满足该stream的条件，可以根据实际情况选择是否实现它。
- SaveDataBolt组件
最后这个Bolt用来模拟保存处理过的数据内容，代码如下：
public static class SaveDataBolt extends BaseRichBolt {
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(SaveDataBolt.class);
private OutputCollector collector;
private String type;
public SaveDataBolt(String type) {
this.type = type;
}
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
// just print the received tuple for being waited to persist
LOG.info("[" + type + "] " +
"SourceComponent=" + input.getSourceComponent() +
", SourceStreamId=" + input.getSourceStreamId() +
", type=" + input.getString(0) +
", value=" + input.getString(1));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// do nothing
}
}
在实际应用中，你可能需要将处理过的数据保存到数据库中，就可以在该Bolt中实现存储数据的逻辑。
总结
Storm中最核心的计算组件的抽象就是Spout、Bolt，以及Stream Grouping，其它高级的功能，像Trident、DRPC，他们或者基于这些基础组件以及Streaming Grouping分发策略来实现的，屏蔽了底层的分发计算处理逻辑以更高层的编程抽象面向开发者，减轻了开发人员对底层复杂机制的处理；或者是为了方便使用Storm计算服务而增加的计算机制衍生物，如批量事务处理、RPC等。
参考链接
相关 [storm 实时计算 编程] 推荐：
【转】使用Storm处理事务型实时计算需求时的几处难点
- - 互联网 - ITeye博客接触流计算领域不长时间，对这个领域可以说还是个门外汉. 最近在做实时计算相关的应用，简单说下自己的感受，以后再展开来讨论. 比流量或者订单淘宝可以把我们甩出几条大街. 淘宝的兄弟可以自豪地说他们的实时应用已经承受住了双十一全世界范围内最大的单日数据流的冲击. 而阿里巴巴中文站的流量和订单与淘宝相比则少的可怜.
storm简介
- - 搜索技术博客－淘宝伴随着信息科技日新月异的发展，信息呈现出爆发式的膨胀，人们获取信息的途径也更加多样、更加便捷，同时对于信息的时效性要求也越来越高. 举个搜索场景中的例子，当一个卖家发布了一条宝贝信息时，他希望的当然是这个宝贝马上就可以被卖家搜索出来、点击、购买啦，相反，如果这个宝贝要等到第二天或者更久才可以被搜出来，估计这个大哥就要骂娘了.
Storm Trident 学习
- - 小火箭Storm支持的三种语义：. 至少一次语义的Topology写法. 参考资料： Storm消息的可靠性保障
Storm提供了Acker的机制来保证数据至少被处理一次，是由编程人员决定是否使用这一特性，要使用这一特性需要：. 在Spout emit时添加一个MsgID，那么ack和fail方法将会被调用当Tuple被正确地处理了或发生了错误.
Storm实战之WordCount
- - 编程语言 - ITeye博客 在全面介绍Storm之前，我们先通过一个简单的Demo让大家整体感受一下什么是Storm. 本地模式(Local Mode)： 即Topology（相当于一个任务，后续会详细讲解） 运行在本地机器的单一JVM上，这个模式主要用来开发、调试. 远程模式(Remote Mode):在这个模式，我们把我们的Topology提交到集群，在这个模式中，Storm的所有组件都是线程安全的，因为它们都会运行在不同的Jvm或物理机器上，这个模式就是正式的生产模式.
storm常见问题解答
- - BlogJava-庄周梦蝶 最近有朋友给我邮件问一些storm的问题，集中解答在这里. 一、我有一个数据文件，或者我有一个系统里面有数据，怎么导入storm做计算. 你需要实现一个Spout，Spout负责将数据emit到storm系统里，交给bolts计算. 怎么实现spout可以参考官方的kestrel spout实现：.
Storm 实时性分析
- - CSDN博客架构设计推荐文章都说Storm是一个实时流处理系统，但Storm的实时性体现在什么方面呢. 首先有一个前提：这里的实时性和我们通常所说的实时系统（芯片+汇编或C编写的实时处理软件）的实时性肯定是没法比的，也不是同一个概念. 这里的实时性应该是一个相对的实时性（相对于Hadoop之类 ）. 总结一下，Storm的实时性可能主要体现在：.
那些storm的坑坑
- - 开源软件 - ITeye博客转载请声明出处：http://blackwing.iteye.com/blog/2147633. 在使用storm的过程中，感觉它还是不如hadoop那么成熟. 当然，它的流式处理能力挺让人眼前一亮，以前做的个性化推荐都是离线计算，现在总算把实时部分也加上了. 总结一下storm使用的些心得：. 1.尽量把大量数据处理行为分拆成多个处理component.
storm准实时应用
- - CSDN博客推荐文章1 应用背景： 需要实时统计用户的登陆数，在线人数，活跃时间，下载等指标的数据，或者清洗后移到hdfs上. 1） 客户端产生数据---. 2） kafka-生产者实时采集数据（保留7天）-----. 3） storm实时消费数据，处理数据. 4）把实时数据统计结果缓存到memcached 中.
Storm核心概念剖析
- - 互联网 - ITeye博客最近团队中有分析的场景，用到了JStorm来做数据的实时分析，于是花时间对于一些概念做了了解. 这个的话出来应该有几年时间了，阿里巴巴也重写了一套JStorm，核心的类名都是服用的Storm的，他是一套实时数据处理系统，容错行好，然后足够稳定，目前很多数据实时分析的场景，选择Storm的越来越多了.