`
youxinrencwx
  • 浏览: 66527 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

Storm 常见使用模式-场景

阅读更多

 

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调用,而不是串行地一次只执行一个。
 
 
 
 
 
 
分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics