[TOC] # 數據樣本 ~~~ i am jdxia i am xjd i am jdxia i am jelly ~~~ # jar包 ~~~ <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <!--<scope>provided</scope>--> <version>0.9.5</version> </dependency> ~~~ 安裝log4j # 數據獲取 ~~~ package com.learnstorm; 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 org.apache.commons.lang.StringUtils; import java.io.*; import java.util.ArrayList; import java.util.List; import java.util.Map; //數據獲取 public class MyLocalFileSpout extends BaseRichSpout { private SpoutOutputCollector collector; private BufferedReader bufferedReader; //初始化方法 @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; try { //定義這個去讀取數據 this.bufferedReader = new BufferedReader(new FileReader(new File("/Users/jdxia/Desktop/MyFile/i.txt"))); } catch (FileNotFoundException e) { e.printStackTrace(); } } //storm流式計算的特征就是數據一條一條的處理 // while(true) { // this.nextTuple(); // } //這個方法會被循環調用 @Override public void nextTuple() { //每被調用一次就會發送一條數據出去 try { //讀取一行 String line = bufferedReader.readLine(); //如果不是空的話 if (StringUtils.isNotBlank(line)) { List<Object> arrayList = new ArrayList<Object>(); //把數據放到ArrayList中 arrayList.add(line); //把數據發出去 collector.emit(arrayList); } } catch (IOException e) { e.printStackTrace(); } } //定義下我的輸出 @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("juzi")); } } ~~~ # 數據截取 ~~~ package com.learnstorm; 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; //相當于map-->world,1 //業務邏輯 //對句子進行切割 public class MySplitBolt extends BaseBasicBolt { //處理函數 @Override public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) { //1.數據如何獲取,用tuple獲取 //強轉為string,juzi是上一步定義的 String juzi = (String) tuple.getValueByField("juzi"); //2.進行切割 String[] strings = juzi.split(" "); //3.發送數據 for (String word : strings) { //我們之前用ArrayList存儲,這邊怎么變為Values //可以看下Values的源碼,他是繼承了ArrayList,他存的時候用了一個循環 basicOutputCollector.emit(new Values(word, 1)); } } //定義下我的輸出 //單詞world和他的次數 @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { outputFieldsDeclarer.declare(new Fields("word", "num")); } } ~~~ # 單詞統計 ~~~ package com.learnstorm; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseBasicBolt; import backtype.storm.tuple.Tuple; import java.util.HashMap; import java.util.Map; //打印 public class MyWordCountAndPrintBolt extends BaseBasicBolt { private Map<String, Integer> wordCountMap = new HashMap<String, Integer>(); //處理函數 @Override public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) { //根據之前定義的word和num //強轉為string String word = (String) tuple.getValueByField("word"); Integer num = (Integer) tuple.getValueByField("num"); //1.查看單詞對應的value是否存在 Integer integer = wordCountMap.get(word); if (integer == null || integer.intValue() == 0) { //如果不存在就直接放入新的 wordCountMap.put(word, num); } else { //如果之前已經有了,就把對應統計加上 wordCountMap.put(word, integer.intValue() + num); } System.out.println(wordCountMap); } //不需要定義輸出字段了 @Override public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { } } ~~~ # 任務描述 這邊寫的是本地提交到集群 ~~~ package com.learnstorm; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.generated.AlreadyAliveException; import backtype.storm.generated.InvalidTopologyException; import backtype.storm.generated.StormTopology; import backtype.storm.topology.TopologyBuilder; public class StormTopologyDriver { public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException { //1. 描述任務 TopologyBuilder topologyBuilder = new TopologyBuilder(); //任務的名字自己定義 topologyBuilder.setSpout("mySpout", new MyLocalFileSpout()); //shuffleGrouping和前一個任務關聯 topologyBuilder.setBolt("bolt1", new MySplitBolt()).shuffleGrouping("mySpout"); topologyBuilder.setBolt("bolt2", new MyWordCountAndPrintBolt()).shuffleGrouping("bolt1"); //2. 任務提交 //提交給誰?提交什么內容? Config config = new Config(); StormTopology stormTopology = topologyBuilder.createTopology(); //本地模式 LocalCluster localCluster = new LocalCluster(); localCluster.submitTopology("wordCount", config, stormTopology); //集群模式 // StormSubmitter.submitTopology("worldCount1", config, stormTopology); } } ~~~