大数据学习之storm-wordcount 实时版开发以及分组策略34

2024-04-07 23:11:51

五：storm-wordcount 实时版开发

1：编写Spout

package wc;

import java.util.Map;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * @author Dawn
 * @date 2019年6月5日15:59:04
 * @version 1.0
 * 需求：单词计数  hello dawn hello dawn indicate
 * 实现接口：IRichSpout	IRichBolt
 * 继承抽象类:BaseRichSpout BaseRichSpout	常用
 */
public class WordCountSpout extends BaseRichSpout{

	//定义收集器
	private SpoutOutputCollector collector;
	
	//发送数据
	@Override
	public void nextTuple() {
		//1:发送数据到blot
		collector.emit(new Values("hello dawn hello dawn indicate"));
		
		//2.设置延迟
		try {
			Thread.sleep(500);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	//创建收集器
	@Override
	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector collector) {

		this.collector=collector;
	}

	//声明
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
		//起别名
		declare.declare(new Fields("dawn"));
	}

}

2：编写分词bolt

package wc;

import java.util.Map;

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.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * @author Dawn
 * @date 2019年6月5日16:09:58
 * @version 1.0
 * 单词切分Bolt组件
 */
public class WordCountSplitBolt extends BaseRichBolt{
	
	//数据继续发送到下一个bolt
	private OutputCollector collector;

	@Override
	public void execute(Tuple in) {
		//1.获取数据
		String line = in.getStringByField("dawn");
		
		//2.切分数据
		String[] fields = line.split(" ");

		//3.<单词,1> 发送出去 下一个bolt(累加求和)
		for(String w:fields) {
			collector.emit(new Values(w,1));
		}
	}

	//初始化
	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
		this.collector=collector;
	}

	//声明描述
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
		declare.declare(new Fields("word","sum"));
	}

}

3：编写计数bolt

package wc;

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

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.Fields;
import org.apache.storm.tuple.Tuple;

/**
 * @author Dawn
 * @date 2019年6月5日16:17:45
 * @version 1.0
 * 计数bolt
 */
public class WordCount extends BaseRichBolt{

	private Map<String, Integer> map=new HashMap<>();
	
	//累加求和
	@Override
	public void execute(Tuple in) {
		//1.获取数据
		String word = in.getStringByField("word");
		Integer sum = in.getIntegerByField("sum");
		
		//2.业务处理
		if(map.containsKey(word)) {
			//之前出现几次
			Integer count = map.get(word);
			//已有的
			map.put(word, count+sum);
		}else {
			map.put(word, sum);
		}
		
		//3.打印控制台
				System.err.println(Thread.currentThread().getName() + "\t 单词为：" + word + "\t 当前已出现次数为：" + map.get(word));
	}

	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector collector) {
	}

	//没有下一个阶段就不用写
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
	}

}

4：编写driver驱动类

package wc;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @author Dawn
 * @date 2019年6月5日16:18:52
 * @version 1.0
 * 驱动类，以及使用不同的分组策略演示(字段，随机，全局)
 */
public class WordCountDriver {

	public static void main(String[] args) {
		//1.hadoop->Job storm->topology 创建拓扑
		TopologyBuilder builder = new TopologyBuilder();
		
		//2.指定设置
		//设置任务的spout组件
		builder.setSpout("WordCountSpout", new WordCountSpout(),2);//拓扑名，数据源，并行度
		
		//设置任务的单词拆分的bolt组件,是字段分组 并行度为2，总任务数 4
		builder.setBolt("WordCountSplitBolt", new WordCountSplitBolt(),2).setNumTasks(4).fieldsGrouping("WordCountSpout", new Fields("dawn"));
				
		//设置任务的单词计数的bolt组件，是字段分组 ,并行度为2
		builder.setBolt("WordCount", new WordCount(),4).fieldsGrouping("WordCountSplitBolt", new Fields("word","sum"));
		
//============================================================================================================》		
//		//设置任务的单词拆分的bolt组件,是随机分组 并行度为2，总任务数 4
//		builder.setBolt("WordCountSplitBolt", new WordCountSplitBolt(),2).setNumTasks(4).shuffleGrouping("WordCountSpout");
//		
//		//设置任务的单词计数的bolt组件，是随机分组 ,并行度为2
//		builder.setBolt("WordCount", new WordCount(),4).shuffleGrouping("WordCountSplitBolt");
		
//============================================================================================================》			
//		//设置任务的单词拆分的bolt组件,是全局分组 并行度为2，总任务数 4
//		//分配给task id值最小的 根据线程id判断，只分噢诶给线程id最小的  
//		builder.setBolt("WordCountSplitBolt", new WordCountSplitBolt(),2).setNumTasks(4).globalGrouping("WordCountSpout");
//				
//		//设置任务的单词计数的bolt组件，是全局分组 ,并行度为2
//		builder.setBolt("WordCount", new WordCount(),4).globalGrouping("WordCountSplitBolt");
//		
		//3.创建配置信息
		Config conf = new Config();
		//conf.setNumWorkers(10);可以设置work数
		
//		//集群模式运行
//		try {
//			StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
//		} catch (AlreadyAliveException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		} catch (InvalidTopologyException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		} catch (AuthorizationException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
		
		//4.提交任务(本地模式)
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("wordcounttopology", conf, builder.createTopology());
		
	}

}