admin 管理员组

文章数量: 1086019

storm

数据流组

设计一个拓扑时，你要做的最重要的事情之一就是定义如何在各组件之间交换数据（数据流是如何被bolts消费的）。一个数据流组指定了每个bolt会消费哪些数据流，以及如何消费它们。

数据流组在定义拓扑时设置，就如

builder.setBolt("当前groupID",  处理该流的Bolt实例, [int 并行数]).shuffleGrouping("指定接收的groupID");

TopologyBuilder对象通过<font color="red’>setBolt方法来规定每个Bolt所属于的group，通过规定group，实现bolt直接数据接收来源以及先后顺序。

**NOTE:**一个节点能够发布一个以上的数据流，一个数据流组允许我们选择接收哪个。

随机数据流组

shuffleGrouping(指定接收的groupID)

随机流组是最常用的数据流组。它只有一个参数（数据源组件），并且数据源会向随机选择的bolt发送元组，保证每个消费者收到近似数量的元组。
随机数据流组用于数学计算这样的原子操作。然而，如果操作不能被随机分配，就要考虑其它分组方式了。

域数据流组

域数据流组允许你基于元组的一个或多个域控制如何把元组发送给bolts。它保证拥有相同域组合的值集发送给同一个bolt。

 builder.setBolt("word-counter", new WordCounter(),2).fieldsGrouping("word-normalizer", new Fields("word"));

如果你用word域为数据流分组，word-normalizer bolt将只会把相同单词的元组发送给同一个word-counter bolt实例。

全部数据流组

全部数据流组，为每个接收数据的实例复制一份元组副本。这种分组方式用于向bolts发送信号。比如，你要刷新缓存，你可以向所有的bolts发送一个刷新缓存信号。

// 拓扑定义一个全部数据流组，该流组的ID定义为signalsbuilder.setBolt("word-counter", new WordCounter(),2).fieldsGroupint("word-normalizer",new Fields("word")).allGrouping("signals-spout","signals");// 在WordCounter()类的execute(Tuple input)方法中，我们可以通过input.getSourceStreamId().equals("signals")来判断这个数据量是否是全部数据流组
public void execute(Tuple input) {if(input.getSourceStreamId().equals("signals")){//刷新操作定义于此}
}

** Storm允许我们声明具名数据流（如果你不把元组发送到一个具名数据流，默认发送到名为”default“的数据流）。 **

提交拓扑到集群

用一个叫做LocalCluster的工具在你的本地机器上运行了一个拓扑。Storm的基础工具，使你能够在自己的计算机上方便的运行和调试不同的拓扑。
要实现提交到真实的集群，你需要把LocalCluster换成StormSubmitter并实现submitTopology方法， 它负责把拓扑发送给集群。

//LocalCluster cluster = new LocalCluster();//cluster.submitTopology("Count-Word-Topology-With-Refresh-Cache", conf, //builder.createTopology());StormSubmitter.submitTopology("Count-Word-Topology-With_Refresh-Cache", conf,builder.createTopology());//Thread.sleep(1000);//cluster.shutdown();```接下来，把源码压缩成一个jar包现在你生成了一个jar包，使用storm jar命令提交拓扑。

本文标签： storm

版权声明：本文标题：storm 内容由网友自发贡献，该文观点仅代表作者本人， 转载请联系作者并注明出处：http://roclinux.cn/b/1687544518a114211.html， 本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容，一经查实，本站将立刻删除。