Storm技术基础（一）

package com.demo.wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;

/**
 * 驱动类（构造拓扑）
 * @author zhangqingli
 *
 */
public class WordcountTopology {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();
		
		//线程数为2
		builder.setSpout("randomSentenceSpout", new RandomSentenceSpout(), 2); 
		//线程数为2，使用随机分组，订阅的消息为randomSentenceSpout
		builder.setBolt("wordNomalizerBolt", new WordNomalizerBolt(), 2).shuffleGrouping("randomSentenceSpout"); 
		//线程数为2，使用按字段分组，订阅的消息为wordNomalizerBolt
		builder.setBolt("wordCountBolt", new WordCountBolt(), 2).fieldsGrouping("wordNomalizerBolt", new Fields("word")); 
		//线程数为1，使用全局分组（慎用，数据量可能很大），订阅的消息为wordCountBolt
		builder.setBolt("printBolt", new PrintBolt(), 1).globalGrouping("wordCountBolt");
		//设置一些参数
		Config stormConf = new Config(); //在spout或bolt中可以通过conf map入参获取
		stormConf.setDebug(false);
		stormConf.setNumAckers(1);
		
		//通过是否有参数来控制是否启动集群（或本地方式运行）
		if(args!=null && args.length>0) {
			//最好一台机器上的一个topology只使用一个worker,主要原因时减少了worker之间的数据传输
			stormConf.setNumWorkers(1);
			try {
				StormSubmitter.submitTopology(args[0], stormConf, builder.createTopology());
			} catch (Exception e) {
				e.printStackTrace();
			}
		} else {
			stormConf.setMaxTaskParallelism(1);
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("wordcount", stormConf, builder.createTopology());
			
			//本地集群需要手动关闭
			Utils.sleep(60000);
			localCluster.killTopology("wordcount");
			localCluster.shutdown();
		}
	}
}

package com.demo.wordcount;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

/**
 * Spout组件（继承BaseRichSpout 或 实现IRichSpout）
 * @author zhangqingli
 *
 */
public class RandomSentenceSpout implements IRichSpout {
	private static final long serialVersionUID = 1L;
	
	private SpoutOutputCollector spoutOutputCollector;
	private Map<String, Values> tupleMap; //tuple缓存器

	/**
	 * 初始化方法
	 * 首先被调用，且只被调用一次，一般用于初始化变量，如SpoutOutputCollector发射器
	 */
	@Override
	@SuppressWarnings("rawtypes")
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { 		  //conf表示驱动类中的stormConf
		spoutOutputCollector = collector;
		tupleMap = new HashMap<>();
	}
	
	/**
	 * 收尾方法
	 * 一般用于释放资源
	 */
	@Override
	public void close() {
		 
	}
	
	/**
	 * 激活方法
	 * 当拓扑任务被暂停重新激活时会调用该方法，一般该方法默认即可
	 */
	@Override
	public void activate() {
		
	}
	
	/**
	 * 暂停方法
	 * 当拓扑任务被暂停时会调用该方法，一般该方法默认即可
	 */
	@Override
	public void deactivate() {
		
	}
	
	/**
	 * 核心方法
	 * 会被循环调用，实现如何从数据源上获取数据和向后面的组件bolt发射数据的逻辑
	 */
	@Override
	public void nextTuple() {
		Utils.sleep(5000);
		String[] sentences = new String[]{
				"Implement a parser wrapper by extending",
				"the FeedNormalizer Parser class and",
				"overriding the public methods. Also note", 
				"the helper methods in the root Parser",
				"object to make mapping of output from", 
				"the particular parser to the Feed object easier."
		};
		String sentence = sentences[new Random().nextInt(sentences.length)];
		
		String msgId = UUID.randomUUID().toString(); //msgId（消息保障机制用）
		Values values = new Values(sentence.trim().toLowerCase()); //tuple
		tupleMap.put(msgId, values); //将tuple添加到自定义缓存中
		spoutOutputCollector.emit(values, msgId); //发射
		
		System.err.println(Thread.currentThread() + "，发射的语句：" + sentence);
		System.err.println(Thread.currentThread() + "，tupleMap缓存：" + tupleMap);
	}
	
	
	/**
	 * 消息可靠性保障ack方法
	 * 当拓扑任务启用了消息可靠性保障机制，当某个tuple在拓扑任务上处理成功后，调用ack方法执行一些处理逻辑
	 * 
	 * 启用消息的确认机制步骤：
	 * 在第一个spout或bolt中：
	 * 		1. 发射器发射tuple时需要指定一个msgId
	 * 		2. 使用map缓存所发射的tuple（key为msgId，value为emit_values）
	 * 		3. 在ack方法中（此时已经能够确认消息发射成功）将缓存从map中移除
	 * 		4. 在fail方法中（此时已经能够确认消息发射失败）进行重试或其他补救操作
	 * 在下一个spout或bolt中：（如果需要继续往后面组件发射消息）
	 * 		1. 需要锚定前面的tuple【collector.emit(inputTuple, new Values("xxx"))】
	 * 		2. 消息处理成功调用收集器的ack方法【collector.ack(inputTuple)】
	 * 		3. 消息处理过程有异常调用收集器的fail方法【collector.fail(inputTuple)】
	 * 
	 * 对性能的影响：
	 * 		如果启用了消息保障性机制，一定会对拓扑任务的执行的效率产生影响，如果性能影响超过可接受范围，
	 * 		解决方法是增加acker组件的并发度（增加执行acker任务的executor线程个数），即可以通过设置
	 * 		stormConf.setNumAckers(5)来做。
	 * 
	 * 关于锚定：
	 * 所谓的锚定(anchoring)是指为 tupleTree 中增加一个新的节点。在bolt中你可以通过 
	 * collector.emit(tuple,new Values(xx)) 将该tuple发送到下个bolt，此时new Value 
	 * 就被锚定在该 tuple 上了。你可以认为 tuple tree 之间link的建立称为锚定。
	 * 
	 */
	@Override
	public void ack(Object msgId) {
		System.err.println(Thread.currentThread() + "，消息发射成功msgId：" + msgId);
		//确认发射成功，将tuple从缓存中移除
		tupleMap.remove(msgId);
	}

	/**
	 * 消息可靠性保障fail方法
	 * 当拓扑任务启用了消息可靠性保障机制，当某个tuple在拓扑任务上处理失败或超时后，
	 * 调用fail方法执行一些处理逻辑
	 * 
	 * 关于成功与失败：
	 * 在Storm的一个Topology中，Spout通过 SpoutOutputCollector 的emit()方法
	 * 发射一个tuple(源)即消息。而后经过在该Topology中定义的多个Bolt处理时，可能
	 * 会产生一个或多个新的Tuple。源Tuple和新产生的Tuple便构成了一个Tuple树。当整
	 * 棵树被处理完成，才算一个Tuple被完全处理，其中任何一个节点的Tuple处理失败或超
	 * 时，则整棵树失败。关于消息在storm中的超时默认为30秒，具体参看 defaults.yaml 
	 * 中的 topology.message.timeout.secs: 30 的配置。同时，你也可以在定义
	 * topology时，通过conf.setMessageTimeoutSecs方法指定超时时间。storm所谓的
	 * 消息可靠性指的是storm保证每个tuple都能被toplology完全处理。而且处理的结果要
	 * 么成功要么失败。出现失败的原因可能有两种即节点处理失败或者处理超时。
	 * 
	 * 有三种方法可以去掉可靠性:
	 * 1. 第一是把Config.TOPOLOGY_ACKERS 设置成 0. 在这种情况下，storm会在spout
	 *    发射一个tuple之后马上调用spout的ack方法。也就是说这个tuple树不会被跟踪。
	 * 2. 第二个方法是在tuple层面去掉可靠性。 你可以在发射tuple的时候不指定messageid
	 *    来达到不跟粽某个特定的spout tuple的目的。
	 * 3. 最后一个方法是如果你对于一个tuple树里面的某一部分到底成不成功不是很关心，那么
	 *    可以在发射这些tuple的时候unanchor它们。这样这些tuple就不在tuple树里面，也
	 *    就不会被跟踪了。
	 */
	@Override
	public void fail(Object msgId) {
		System.err.println(Thread.currentThread() + "，消息发射失败msgId：" + msgId);
		//消息发射失败，重试
		Values values = tupleMap.get(msgId);
		spoutOutputCollector.emit(values, msgId);
	}

	/**
	 * 声明向后面发射的tuple的key
	 * 
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("sentence"));
	}

	/**
	 * 设置spout组件专用的参数
	 */
	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
}

package com.demo.wordcount;

import java.util.Map;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class WordNomalizerBolt implements IRichBolt {
	private static final long serialVersionUID = 1L;

	private OutputCollector outputCollector;
	
	/**
	 * worker启动时初始化方法
	 * 功能类似于spout的open
	 */
	@Override
	@SuppressWarnings("rawtypes")
	public void prepare(Map paramMap, TopologyContext paramTopologyContext,
			OutputCollector paramOutputCollector) {
		outputCollector = paramOutputCollector;
	}

	/**
	 * 接收一个tuple并执行逻辑处理，发射出去
	 */
	@Override
	public void execute(Tuple paramTuple) {
		
		try {
			String sentence = paramTuple.getStringByField("sentence");
			String[] words = sentence.split(" ");
			
			for (String word : words) {
				//outputCollector.emit(new Values(word));
                //启用消息保障机制，需要锚定接收到的tuple
				outputCollector.emit(paramTuple, new Values(word)); 
			}
			
			//确认消息处理成功
			outputCollector.ack(paramTuple);
		} catch (Exception e) {
			e.printStackTrace();
			//确认消息处理失败
			outputCollector.fail(paramTuple); 
		}
	}

	/**
	 * 收尾方法（不保证一定被调用，因为我们停止拓扑任务的时候一般是采取杀进程的方法）
	 * 一般用于释放资源
	 */
	@Override
	public void cleanup() {
		
	}

	/**
	 * 声明向后面发射的tuple的key
	 * 
	 */
	@Override
	public void declareOutputFields(
			OutputFieldsDeclarer paramOutputFieldsDeclarer) {
		paramOutputFieldsDeclarer.declare(new Fields("word"));
	}

	/**
	 * 设置spout组件专用的参数
	 */
	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}
}

package com.demo.wordcount;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class WordCountBolt implements IRichBolt {
	private static final long serialVersionUID = 1L;
	private OutputCollector outputCollector;
	private Map<String, Integer> counterMap;
	
	@Override
	public void cleanup() {
		
	}

	@Override
	public void execute(Tuple inputTuple) {
		String str = inputTuple.getStringByField("word");
		
		if (!counterMap.containsKey(str)) {
			counterMap.put(str, 1);
		} else {
			Integer c = counterMap.get(str) + 1;
			counterMap.put(str, c);
		}
		
		//只取词频最大的前十个
		int num = 10;
		int length = 0;
		
		//对map进行排序
		counterMap = sortByMap(counterMap);
		
		if (counterMap.size() > num) {
			length = num;
		} else {
			length = counterMap.size();
		}
		
		//增量统计
		String countResult = null;
		int count = 0;
		for (String key : counterMap.keySet()) {
			if (count >= length) {
				break;
			}
			if (count==0) {
				countResult = "["+ key + ":" + counterMap.get(key) +"]";
			} else {
				countResult = countResult + ", ["+ key + ":" + counterMap.get(key) +"]";
			}
			count++;
		}
		
		countResult = "The first " + num + ":" + countResult;
        //该部分不启用消息保障机制，不需要锚定接收到的tuple
		outputCollector.emit(new Values(countResult)); 
		outputCollector.ack(inputTuple);
	}

	@Override
	@SuppressWarnings("rawtypes")
	public void prepare(Map stormConf, TopologyContext context, 
                        OutputCollector collector) {
		outputCollector = collector;
		counterMap = new HashMap<String, Integer>();
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("countResult"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	/*
	 * 根据map的value对map进行排序
	 */
	public Map<String, Integer> sortByMap(Map<String, Integer> map) {
		List<Entry<String, Integer>> list = 
          	new LinkedList<Entry<String, Integer>>(map.entrySet());
		Collections.sort(list, new Comparator<Entry<String, Integer>>() {
			@Override
			public int compare(Entry<String, Integer> o1,
					Entry<String, Integer> o2) {
				return o2.getValue().compareTo(o1.getValue());
			}
		});
		Map<String, Integer> result = new LinkedHashMap<String, Integer>();
		for (Entry<String, Integer> entry : list) {
			result.put(entry.getKey(), entry.getValue());
		}
		return result;
	}
}

package com.demo.wordcount;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

public class PrintBolt extends BaseBasicBolt {
	private static final long serialVersionUID = 1L;
	
	@Override
	public void execute(Tuple tuple, BasicOutputCollector collector) {
		String msg = tuple.getString(0);
		if (msg != null) {
			System.out.println(Thread.currentThread() + "，" + msg);
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));
	}
}

// 可以通过stormConfig来设置worker进程数(默认为1个)
// 一般情况下，最好一台机器上的一个topology只使用一个worker,主要原因时减少了worker之间的数据传输
stormConfig.setNumWorkers(2)

TopologyBuilder builder = new TopologyBuilder();
//线程数为2
builder.setSpout("randomSentenceSpout", new RandomSentenceSpout(), 2); 

//线程数为2，使用随机分组，订阅的消息为randomSentenceSpout
builder.setBolt("wordNomalizerBolt", 
                new WordNomalizerBolt(), 2).shuffleGrouping("randomSentenceSpout");

//表示该bolt有12个task来运行，12个task由3个executor线程来运行
//相当于每个executor线程要运行4个task，这4个task只能executor线程上轮流执行
builder.setBolt("wordNomalizerBolt", 
                new WordNomalizerBolt(), 3)
  				.setNumTasks(12)
  				.shuffleGrouping("randomSentenceSpout");