您的位置:首页 > 大数据

大数据入门第十六天——流式计算之storm详解(二)常用命令与wc实例

2018-03-06 13:52 671 查看

一、常用命令

  1.提交命令

提交任务命令格式:storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】
torm jar examples/storm-starter/storm-starter-topologies-0.9.6.jar storm.starter.WordCountTopology wordcount


  2.杀死任务

storm kill 【拓扑名称】 -w 10(执行kill命令时可以通过-w [等待秒数]指定拓扑停用以后的等待时间)
storm kill topology-name -w 10


  3.停用任务

storm deactivte  【拓扑名称】
storm deactivte topology-name
#我们能够挂起或停用运行中的拓扑。当停用拓扑时,所有已分发的元组都会得到处理,但是spouts的nextTuple方法不会被调用。
销毁一个拓扑,可以使用kill命令。它会以一种安全的方式销毁一个拓扑,首先停用拓扑,在等待拓扑消息的时间段内允许拓扑完成当前的数据流。


  4.启用任务

storm activate【拓扑名称】
storm activate topology-name


  5.重新部署任务

storm rebalance  【拓扑名称】
storm rebalance topology-name
再平衡使你重分配集群任务。这是个很强大的命令。比如,你向一个运行中的集群增加了节点。
再平衡命令将会停用拓扑,然后在相应超时时间之后重分配工人,并重启拓扑。


  更多命令,参考:http://blog.csdn.net/u010003835/article/details/52123807

  完整命令,参考官网文档:http://storm.apache.org/releases/1.0.6/Command-line-client.html

二、wordCount示例程序

 ###以下内容应当替换为Jstrom的依赖!包是backtype的topologyBuilder!

   1.引入依赖

<!-- https://mvnrepository.com/artifact/org.apache.storm/storm-core -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.0.6</version>
<scope>provided</scope>
</dependency>


  JStorm是阿里巴巴基于storm的二次开发,完全兼容storm!

<!-- https://mvnrepository.com/artifact/com.alibaba.jstorm/jstorm-core -->
<dependency>
<groupId>com.alibaba.jstorm</groupId>
<artifactId>jstorm-core</artifactId>
<version>2.1.1</version>
<!--<scope>provided</scope>-->
</dependency>


  // 本地提交时请注释掉作用域(provided不参与打包)

  2。编写相关程序

参考:http://blog.csdn.net/u010454030/article/details/52576346

    http://m635674608.iteye.com/blog/2221179

package com.jiangbei;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
* 自定义bolt
*
* @author zcc ON 2018/3/6
**/
public class MyBolt2 extends BaseRichBolt{
private OutputCollector collector;
private Map<String, Integer> map = new HashMap<>();
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
this.collector = outputCollector;
}

@Override
public void execute(Tuple tuple) {
String word = tuple.getString(0);
Integer num = tuple.getInteger(1);
if (map.containsKey(word)) {
Integer count = map.get(word);
map.put(word, count + num);
} else {
map.put(word, 1);
}
System.out.println("count==========>" + map);
}

@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

}
}


MyBolt2
  本地运行直接运行即可!

  以上的spout、bolt等都是采用的自定义的,主要通过继承BaseRichSpout等来实现!;这里解释一下 其中的一些概念与相关类:

    最后一个bolt必须按照field进行分组,这样才能进行计数!,这里的new Field()里的值必须是上游里面declare的值。才能对应上。

    SpoutOutputCollector——对象提供了发射tuple的方法

  整个过程原理图如下所示:

  


  3.Stream Grouping详解


Storm里面有7种类型的stream grouping

l Shuffle Grouping: 随机分组, 随机派发stream里面的tuple,保证每个bolt接收到的tuple数目大致相同。(类似MR的hash算法)

l Fields Grouping:按字段分组,比如按userid来分组,具有同样userid的tuple会被分到相同的Bolts里的一个task,而不同的userid则会被分配到不同的bolts里的task。

l All Grouping:广播发送,对于每一个tuple,所有的bolts都会收到。

l Global Grouping:全局分组, 这个tuple被分配到storm中的一个bolt的其中一个task。再具体一点就是分配给id值最低的那个task。

l Non Grouping:不分组,这stream grouping个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果, 有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。

l Direct Grouping: 直接分组, 这是一种比较特别的分组方法,用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的task的id (OutputCollector.emit方法也会返回task的id)。

l Local or shuffle grouping:如果目标bolt有一个或者多个task在同一个工作进程中,tuple将会被随机发生给这些tasks。否则,和普通的Shuffle Grouping行为一致。


  相关的图解,参考https://www.cnblogs.com/kqdongnanf/p/4634607.html
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: