[大数据]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定义为signals
 builder.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命令提交拓扑。