單詞計數拓撲WordCountTopology實現的基本功能就是不停地讀入一個個句子，最後輸出每個單詞和數目並在終端不斷的更新結果，拓撲的資料流如下：

• 語句輸入Spout:  從資料來源不停地讀入資料，並生成一個個句子，輸出的tuple格式：{"sentence":"hello world"}
• 語句分割Bolt: 將一個句子分割成一個個單詞，輸出的tuple格式：{"word":"hello"}  {"word":"world"}
• 單詞計數Bolt: 儲存每個單詞出現的次數，每接到上游一個tuple後，將對應的單詞加1，並將該單詞和次數傳送到下游去，輸出的tuple格式：{"hello":"1"}  {"world":"3"}
• 結果上報Bolt: 維護一份所有單詞計數表，每接到上游一個tuple後，更新表中的計數資料，並在終端將結果打印出來。

　 開發步驟：

    1.環境

• 作業系統：mac os 10.10.3
• JDK: jdk1.8.0_40
• IDE: intellij idea 15.0.3
• Maven: apache-maven-3.0.3

　 2.專案搭建

• 在idea新建一個maven專案工程：storm-learning
• 修改pom.xml檔案,加入strom核心的依賴，配置slf4j依賴，方便Log輸出

<dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.6.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.0.2</version>
        </dependency>
</dependencies>

　3. Spout和Bolt元件的開發

• SentenceSpout
• SplitSentenceBolt
• WordCountBolt
• ReportBolt

SentenceSpout.java

 1 public class SentenceSpout extends BaseRichSpout{
 2 
 3     private SpoutOutputCollector spoutOutputCollector;
 4 
 5     //為了簡單,定義一個靜態資料模擬不斷的資料流產生
 6     private static final String[] sentences={
 7
 
             "The logic for a realtime application is packaged into a Storm topology",
 8             "A Storm topology is analogous to a MapReduce job",
 9             "One key difference is that a MapReduce job eventually finishes whereas a topology runs forever",
10             " A topology is a graph of spouts and bolts that are connected with stream groupings"
11     };
12 
13     private int index=0;
14 
15     //初始化操作
16     public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
17         this.spoutOutputCollector = spoutOutputCollector;
18     }
19 
20     //核心邏輯
21     public void nextTuple() {
22         spoutOutputCollector.emit(new Values(sentences[index]));
23         ++index;
24         if(index>=sentences.length){
25             index=0;
26         }
27     }
28 
29     //向下遊輸出
30     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
31         outputFieldsDeclarer.declare(new Fields("sentences"));
32     }
33 }

SplitSentenceBolt.java

 1 public class SplitSentenceBolt extends BaseRichBolt{
 2 
 3     private OutputCollector outputCollector;
 4 
 5     public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
 6         this.outputCollector = outputCollector;
 7     }
 8 
 9     public void execute(Tuple tuple) {
10         String sentence = tuple.getStringByField("sentences");
11         String[] words = sentence.split(" ");
12         for(String word : words){
13             outputCollector.emit(new Values(word));
14         }
15     }
16 
17     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
18         outputFieldsDeclarer.declare(new Fields("word"));
19     }
20 }

WordCountBolt.java

 1 public class WordCountBolt extends BaseRichBolt{
 2 
 3     //儲存單詞計數
 4     private Map<String,Long> wordCount = null;
 5 
 6     private OutputCollector outputCollector;
 7 
 8     public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
 9         this.outputCollector = outputCollector;
10         wordCount = new HashMap<String, Long>();
11     }
12 
13     public void execute(Tuple tuple) {
14         String word = tuple.getStringByField("word");
15         Long count = wordCount.get(word);
16         if(count == null){
17             count = 0L;
18         }
19         ++count;
20         wordCount.put(word,count);
21         outputCollector.emit(new Values(word,count));
22     }
23 
24 
25     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
26         outputFieldsDeclarer.declare(new Fields("word","count"));
27     }
28 }

ReportBolt.java

 1 public class ReportBolt extends BaseRichBolt {
 2     
 3     private static final Logger log = LoggerFactory.getLogger(ReportBolt.class);
 4 
 5     private Map<String, Long> counts = null;
 6 
 7     public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
 8         counts = new HashMap<String, Long>();
 9     }
10 
11     public void execute(Tuple tuple) {
12         String word = tuple.getStringByField("word");
13         Long count = tuple.getLongByField("count");
14         counts.put(word, count);
15         //列印更新後的結果
16         printReport();
17     }
18 
19     public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
20         //無下游輸出,不需要程式碼
21     }
22 
23     //主要用於將結果打印出來,便於觀察
24     private void printReport(){
25         log.info("--------------------------begin-------------------");
26         Set<String> words = counts.keySet();
27         for(String word : words){
28             log.info("@[email protected]: " + word + " ---> " + counts.get(word));
29         }
30         log.info("--------------------------end---------------------");
31     }
32 }

　4.拓撲配置

• WordCountTopology

 1 public class WordCountTopology {
 2 
 3     private static final Logger log = LoggerFactory.getLogger(WordCountTopology.class);
 4 
 5     //各個元件名字的唯一標識
 6     private final static String SENTENCE_SPOUT_ID = "sentence-spout";
 7     private final static String SPLIT_SENTENCE_BOLT_ID = "split-bolt";
 8     private final static String WORD_COUNT_BOLT_ID = "count-bolt";
 9     private final static String REPORT_BOLT_ID = "report-bolt";
10 
11     //拓撲名稱
12     private final static String TOPOLOGY_NAME = "word-count-topology";
13 
14     public static void main(String[] args) {
15 
16         log.info(".........begining.......");
17         //各個元件的例項
18         SentenceSpout sentenceSpout = new SentenceSpout();
19         SplitSentenceBolt splitSentenceBolt = new SplitSentenceBolt();
20         WordCountBolt wordCountBolt = new WordCountBolt();
21         ReportBolt reportBolt = new ReportBolt();
22 
23         //構建一個拓撲Builder
24         TopologyBuilder topologyBuilder = new TopologyBuilder();
25 
26         //配置第一個元件sentenceSpout
27         topologyBuilder.setSpout(SENTENCE_SPOUT_ID, sentenceSpout, 2);
28 
29         //配置第二個元件splitSentenceBolt,上游為sentenceSpout,tuple分組方式為隨機分組shuffleGrouping
30         topologyBuilder.setBolt(SPLIT_SENTENCE_BOLT_ID, splitSentenceBolt).shuffleGrouping(SENTENCE_SPOUT_ID);
31 
32         //配置第三個元件wordCountBolt,上游為splitSentenceBolt,tuple分組方式為fieldsGrouping,同一個單詞將進入同一個task中(bolt例項)
33         topologyBuilder.setBolt(WORD_COUNT_BOLT_ID, wordCountBolt).fieldsGrouping(SPLIT_SENTENCE_BOLT_ID, new Fields("word"));
34 
35         //配置最後一個元件reportBolt,上游為wordCountBolt,tuple分組方式為globalGrouping,即所有的tuple都進入這一個task中
36         topologyBuilder.setBolt(REPORT_BOLT_ID, reportBolt).globalGrouping(WORD_COUNT_BOLT_ID);
37 
38         Config config = new Config();
39 
40         //建立本地叢集,利用LocalCluster,storm在程式啟動時會在本地自動建立一個叢集,不需要使用者自己再搭建,方便本地開發和debug
41         LocalCluster cluster = new LocalCluster();
42 
43         //建立拓撲例項,並提交到本地叢集進行執行
44         cluster.submitTopology(TOPOLOGY_NAME, config, topologyBuilder.createTopology());
45     }
46 }

  5.拓撲執行

• 方法一：通過IDEA執行

　　在idea中對程式碼進行編譯compile，然後run；

　　觀察控制檯輸出會發現，storm首先在本地自動建立了執行環境，即啟動了zookepeer,接著啟動nimbus,supervisor；然後nimbus將提交的topology進行分發到supervisor，supervisor啟動woker程序，woker程序裡利用Executor來執行topology的元件(spout和bolt);最後在控制檯發現不斷的輸出單詞計數的結果。

     zookepeer的連線建立

　  nimbus啟動

　  supervisor啟動

　  worker啟動

     Executor啟動執行

     結果輸出

• 方法二：通過maven來執行
  • 進入到該專案的主目錄下:storm-learning
  • mvn compile 進行程式碼編譯，保證程式碼編譯通過
  • 通過mvn執行程式：

    mvn exec:java -Dexec.mainClass="wordCount.WordCountTopology"

  • 控制檯輸出的結果跟方法一一致

其他資料：