Storm 常见使用模式-场景

转：http://www.youxinrencwx.com/myblog/blog/java/framework/storm/storm-common-usage-pattern-scene/

Strom常见使用模式如下：

1. 流聚合(stream join)
2. 批处理(Batching)
3. BasicBolt
4. 内存内缓存 + fields grouping 组合
5. 计算top N
6. 分布式RPC: CoordinatedBolt和KeyedFairBolt
7. 用TimeCacheMap来高效地保存一个最近被更新的对象的缓存（目前已被申明为Deprecated，其实guava工具包也提供了类似功能 ）

流聚合(stream join)

流聚合是指基于共同的tuple字段把两个或者多个数据流聚合成一个数据流。流聚合比较类似于table join，只是table join的输入在一次执行中是有限的，并且join的语义（条件）是非常明确的。而流聚合的语义是不明确的并且输入流是无限的。

流类型的聚合类型跟具体的应用是有关了，具体的聚合逻辑就需要自己来实现了。例如,，一些场景是把多个流发出的所有的tuple都聚合起来 — 不管多长时间；而另外一些场景则只会聚合一些特定的tuple。而一般聚合都是通过某些字段来进行聚合， 在storm里面可以用fields grouping在相同字段上进行grouping就可以了，比如:

 

builder.setBolt ( "join" , new MyJoiner (), parallelism )

. fieldsGrouping ( "1" , new Fields ( "joinfield1" , "joinfield2" ))

. fieldsGrouping ( "2" , new Fields ( "joinfield1" , "joinfield2" ))

. fieldsGrouping ( "3" , new Fields ( "joinfield1" , "joinfield2" ));

当然，不同的数据流的“相同”字段可以有不一样的名字。

 

批处理(Batching)

有时候为了性能等原因， 你可能想批量处理tuple一起处理。例如，批量把数据更新数据库。

你可以先保存tuple对象的应用，当达到批量处理条件时，在批量处理，处理完后在ack已处理的tuple

 

BasicBolt

很多bolt有些类似的模式:

读一个输入tuple，根据这个输入tuple发射一个或者多个tuple，在execute的方法的最后ack那个输入tuple

遵循这类模式的bolt一般是函数或者是过滤器, 这种模式非常常见，storm为这类模式单独封装了一个接口:IBasicBolt

 

 

内存内缓存 + fields grouping 组合

在bolt的内存里面缓存一些东西非常常见。缓存在和fields grouping结合起来之后就更有用了。比如，你有一个bolt把短链接变成长链接(bit.ly, t.co之类的)。你可以把短链接到长链接的对应关系利用LRU算法缓存在内存里面以避免重复计算。比如组件一发射短链接，组件二把短链接转化成长链接并缓存在内存里面。看一下下面两段代码有什么不一样：

code one：

builder.setBolt("expand", new ExpandUrl(), parallelism)
.shuffleGrouping(1);

code two：

builder.setBolt("expand", new ExpandUrl(), parallelism)
.fieldsGrouping("urls", new Fields("url"));

第二种方式的缓存会比第一种方式的缓存的效率高很多，因为同样的短链接始终被发到同一个task。这会避免不同的机器上有同样的缓存 — 浪费内存， 同时也使得同样的短域名更可能在内存里面找到缓存。

 

 

计算top N

storm的一个常见的持续计算的模式叫做: “streaming top N”。

比如你有一个bolt发射这样的tuple: ["value", "count"]并且你想一个bolt基于这些信息算出top N的tuple。最简单的办法是有一个bolt可以做一个全局的grouping的动作并且在内存里面保持这top N的值。

这个方式对于大数据量的流显然是没有扩展性的， 因为所有的数据会被发到同一台机器， 单机的处理能力始终是有极限的。一个更好的方法是在多台机器上面并行的计算这个流每一部分的top N, 然后再有一个bolt合并这些机器上面所算出来的top N以算出最后的top N(Map Reduce的思想), 代码大概是这样的:

builder.setBolt("rank", new RankObjects(), parallellism)
.fieldsGrouping("objects", new Fields("value"));
builder.setBolt("merge", new MergeObjects())
.globalGrouping("rank");

 

分布式RPC: CoordinatedBolt和KeyedFairBolt

用storm做分布式RPC应用时 CoordinatedBolt和KeyedFairBolt会被用到，它们是storm api自带的模式。

CoordinatedBolt包装你的bolt，确定是不是所有的tuple都处理完了，然后返回结果。它主要使用Direct Stream来实现这种效果。

KeyedFairBolt同样包装你的bolt逻辑，使你的topology可同时处理多个DRPC调用，而不是串行地一次只执行一个。