前言：阅读笔记

storm jar all-my-code.jar backtype.storm.MyTopology arg1 arg2

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvc2ltb25jaGk=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvc2ltb25jaGk=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

下载安装

http://storm.apache.org/downloads.html

1、依赖安装

2、zk集群

http://blog.csdn.net/simonchi/article/details/43019401

3、zeromq&jzmq

4、python

配置执行

storm.yaml

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvc2ltb25jaGk=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

单词计数程序

Spout

package com.cmcc.chiwei.storm;

import java.io.BufferedReader;

import java.io.FileNotFoundException;

import java.io.FileReader;

import java.util.Map;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseRichSpout;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

public class WordReader extends BaseRichSpout {

	private static final long serialVersionUID = 1L;

	private SpoutOutputCollector collector;

	private FileReader fileReader;

	private boolean completed = false;

	public boolean isDistributed() {

		return false;

	}

	public void ack(Object msgId) {

		System.out.println("OK:"+msgId);

	}

	public void close() {}

	public void fail(Object msgId) {

		System.out.println("FAIL:"+msgId);

	}

	/**

	 * The only thing that the methods will do It is emit each

	 * file line

	 */

	public void nextTuple() {

		if(completed){

			try {

				Thread.sleep(1000);

			} catch (InterruptedException e) {

				//Do nothing

			}

			return;

		}

		String str;

		//Open the reader

		BufferedReader reader = new BufferedReader(fileReader);

		try{

			//Read all lines

			while((str = reader.readLine()) != null){

				/**

				 * By each line emmit a new value with the line as a their

				 */

				this.collector.emit(new Values(str),str);

			}

		}catch(Exception e){

			throw new RuntimeException("Error reading tuple",e);

		}finally{

			completed = true;

		}

	}

	/**

	 * We will create the file and get the collector object

	 */

	public void open(Map conf, TopologyContext context,

			SpoutOutputCollector collector) {

		try {

			this.fileReader = new FileReader(conf.get("wordsFile").toString());

		} catch (FileNotFoundException e) {

			throw new RuntimeException("Error reading file ["+conf.get("wordFile")+"]");

		}

		this.collector = collector;

	}

	/**

	 * Declare the output field "word"

	 */

	public void declareOutputFields(OutputFieldsDeclarer declarer) {

		declarer.declare(new Fields("line"));

	}

}

Bolt1

package com.cmcc.chiwei.storm;

import backtype.storm.topology.BasicOutputCollector;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseBasicBolt;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Tuple;

import backtype.storm.tuple.Values;

public class WordNormalizer extends BaseBasicBolt {

	private static final long serialVersionUID = 1L;

	public void cleanup() {}

	/**

	 * The bolt will receive the line from the

	 * words file and process it to Normalize this line

	 *

	 * The normalize will be put the words in lower case

	 * and split the line to get all words in this

	 */

	public void execute(Tuple input, BasicOutputCollector collector) {

        String sentence = input.getString(0);

        String[] words = sentence.split(" ");

        for(String word : words){

            word = word.trim();

            if(!word.isEmpty()){

                word = word.toLowerCase();

                collector.emit(new Values(word));

            }

        }

	}

	/**

	 * The bolt will only emit the field "word"

	 */

	public void declareOutputFields(OutputFieldsDeclarer declarer) {

		declarer.declare(new Fields("word"));

	}

}

Bolt2

package com.cmcc.chiwei.storm;

import java.util.HashMap;

import java.util.Map;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.BasicOutputCollector;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.topology.base.BaseBasicBolt;

import backtype.storm.tuple.Tuple;

public class WordCounter extends BaseBasicBolt {

	private static final long serialVersionUID = 1L;

	Integer id;

	String name;

	Map<String, Integer> counters;

	/**

	 * At the end of the spout (when the cluster is shutdown

	 * We will show the word counters

	 */

	@Override

	public void cleanup() {

		System.out.println("-- Word Counter ["+name+"-"+id+"] --");

		for(Map.Entry<String, Integer> entry : counters.entrySet()){

			System.out.println(entry.getKey()+": "+entry.getValue());

		}

	}

	/**

	 * On create

	 */

	@Override

	public void prepare(Map stormConf, TopologyContext context) {

		this.counters = new HashMap<String, Integer>();

		this.name = context.getThisComponentId();

		this.id = context.getThisTaskId();

	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {}

	public void execute(Tuple input, BasicOutputCollector collector) {

		String str = input.getString(0);

		if(!counters.containsKey(str)){

			counters.put(str, 1);

		}else{

			Integer c = counters.get(str) + 1;

			counters.put(str, c);

		}

	}

}

Topology

package com.cmcc.chiwei.storm;

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.topology.TopologyBuilder;

import backtype.storm.tuple.Fields;

public class TopologyMain {

	public static void main(String[] args) throws InterruptedException {

        //Topology创建拓扑，安排storm各个节点以及它们交换数据的方式

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("word-reader",new WordReader());

		builder.setBolt("word-normalizer", new WordNormalizer())

			.shuffleGrouping("word-reader");

		builder.setBolt("word-counter", new WordCounter(),1)

			.fieldsGrouping("word-normalizer", new Fields("word"));

        //Configuration

		Config conf = new Config();

		conf.put("wordsFile", args[0]);

		conf.setDebug(false);

        //Topology run

		conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);

		LocalCluster cluster = new LocalCluster();

		cluster.submitTopology("Getting-Started-Topology", conf, builder.createTopology());

		Thread.sleep(2000);

		cluster.shutdown();

	}

}

words.txt

hello

world

storm flume hadoop hdfs

what's wrong flume ?

what's up hdfs ?

Hi,storm,what are you doing ?

执行结果

OK:hello

OK:world

OK:storm flume hadoop hdfs

OK:what's wrong flume ?

OK:what's up hdfs ?

OK:Hi,storm,what are you doing ?

-- Word Counter [word-counter-2] --

what's: 2

flume: 2

hdfs: 2

you: 1

storm: 1

up: 1

hello: 1

hadoop: 1

hi,storm,what: 1

are: 1

doing: 1

wrong: 1

?

: 3

world: 1

分析内容：

Spout

open --> nextTuple



Bolt1

declareOutputFields --> execute



Bolt2

prepare --> execute --> cleanup

更具体的内容，将在兴许慢慢解说，我也在研究中。

。。

。。

望各位不吝不吝赐教！。