通過前面6個章節，我們大致瞭解apache storm的核心細節了，現在我們開始寫一些簡單的程式碼，來感受下storm的魅力。

場景——移動呼叫日誌分析

行動電話呼叫號及其持續時間將作為Apache stormd的輸入流,storm將根據撥號方和接收方之間的電話號碼以及通話次數進行分組。

Spout Creation

Spout是storm用於資料生成的一個元件,。

通常,Spout會實現一個IRichSpout介面。

IRichSpout介面有以下重要的方法：

1.open：−提供Spout以及spout的執行環境。executors會執行這個方法來初始化spout。

2.nextTuple

：通過收集器傳送產生的資料。

3.close−關閉Spout時呼叫close方法。

4.declareOutputFields：宣告輸出元組的schema。

5.ack：處理特定的元組

6.fail：指定一個特定的不用處理和再加工元組。

open

open方法簽名如下：

open(map conf,TopologyContext context,SpoutOutputCollectorcollector)

引數解析：

conf：為Spout提供storm配置。

context:在topology中提供Spout的完整資訊，包括：任務id,輸入輸出資訊。

collector:保證我們傳送的資料能被

bolt處理。

nextTuple

nextTuple方法簽名如下：

nextTuple()

nextTuple()定期方法定期的被相同迴圈中的ack()方法和fail()方法呼叫。

當沒有工作要做的時候必須釋放執行緒，以保證其他方法有機會被呼叫。

因此，nextTuple首先要檢查處理是否已經完成。

如果完成，在結果返回之前,為了降低處理器的負載，該執行緒應該至少睡眠1毫秒。

close

Close方法簽名如下：

close()

declareOutputFields

declareOutputFields方法前面如下：

declareOutputFields(OutputFieldsDeclarer declarer)

引數說明：

Declarer：宣告輸出流的ids,輸出欄位,等等。

這個方法用於指定tuple輸出的shema。

ack

ack方法的簽名如下：

ack(Object msgId)

這個方法表明指定的tuple已經被處理過。

fail

fail方法簽名如下：

fail(Object msgId)

表明spout傳送出的資料並沒有被完全處理，storm會重新處理這個資料。

FakeCallLogReaderSpout

現在我們要收集手機日誌的詳細資訊，包含：

1.主叫號碼

2.被叫號碼

3.通話時長

因為我們沒有實時的通話記錄資訊,那麼我們就自己模擬通話記錄。

Random隨機類產生模擬的通話資訊。

完整的程式程式碼如下所示。

Coding − FakeCallLogReaderSpout.java

import java.util.*;

//import storm tuple packages

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

//import Spout interface packages

import backtype.storm.topology.IRichSpout;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;

//Create a class FakeLogReaderSpout which implement IRichSpout interface

   to access functionalities

public class FakeCallLogReaderSpout implements IRichSpout {

   //Create instance for SpoutOutputCollector which passes tuples to bolt.

   private SpoutOutputCollector collector;

   private boolean completed = false;

   //Create instance for TopologyContext which contains topology data.

   private TopologyContext context;

   //Create instance for Random class.

   private Random randomGenerator = new Random();

   private Integer idx = 0;

   @Override

   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {

      this.context = context;

      this.collector = collector;

   }

   @Override

   public void nextTuple() {

      if(this.idx <= 1000) {

         List<String> mobileNumbers = new ArrayList<String>();

         mobileNumbers.add("1234123401");

         mobileNumbers.add("1234123402");

         mobileNumbers.add("1234123403");

         mobileNumbers.add("1234123404");

         Integer localIdx = 0;

         while(localIdx++ < 100 && this.idx++ < 1000) {

            String fromMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));

            String toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));

            while(fromMobileNumber == toMobileNumber) {

               toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));

            }

            Integer duration = randomGenerator.nextInt(60);

            this.collector.emit(new Values(fromMobileNumber, toMobileNumber, duration));

         }

      }

   }

   @Override

   public void declareOutputFields(OutputFieldsDeclarer declarer) {

      declarer.declare(new Fields("from", "to", "duration"));

   }

   //Override all the interface methods

   @Override

   public void close() {}

   public boolean isDistributed() {

      return false;

   }

   @Override

   public void activate() {}

   @Override

   public void deactivate() {}

   @Override

   public void ack(Object msgId) {}

   @Override

   public void fail(Object msgId) {}

   @Override

   public Map<String, Object> getComponentConfiguration() {

      return null;

   }

}

Bolt Creation

Bolt是一個以元組作為輸入，處理元組後產生新的元組作為輸出的元件。

Bolt通常需要實現IRichBolt介面。

在這個程式中,兩個bolt類CallLogCreatorBolt和CallLogCounterBolt被用來執行的操作處理。

IRichBolt介面有如下方法:

1.prepare ：準備−提供bolt以及bolt的執行環境。

executors會執行這個方法去初始化bolt。

2.execute：處理輸入的單個tuple。

3.cleanup ：要關閉bolt時被呼叫。

4.declareOutputFields：宣告輸出元組的schema。

Prepare

prepare方法的簽名如下:

prepare(Map conf, TopologyContext context, OutputCollector collector)

引數說明：

Conf：為bolt提供配置

Context:在topology中提供完整的bolt位置資訊,包括它的任務id,輸入和輸出資訊等。

Collector:保證處理過的tuple能被髮送出去。

Execute

execute方法簽名如下：

execute(Tuple tuple)//這裡的tuple是將被處理的輸入tuple

execute方法每次處理一個元組。

通過tuple的getValue方法訪問元組的資料。

輸入元組不是必須立即處理的，可以過一會再處理。

可以處理多個元組，處理後產生單個tuple作為輸出tuple。

處理過的tuple可以使用OutputCollector類傳送出去。

Cleanup

cleanup方法簽名如下：

Cleanup()

declareOutputFields

declareOutputFields方法簽名如下：

declareOutputFields(OutputFieldsDeclarer declarer)//這裡的declarer用來宣告輸出流的ids，輸出欄位等資訊

這個方法用來指定tuple的輸出shema。

Call log Creator Bolt

Call log creator bolt 接收通話日誌tuple。

通話日誌tuple包含主叫號碼，被叫號碼和通話時長。

This bolt simply creates a new value by combining the caller number and the receiver number.

這個螺栓簡單地建立了一個新值通過呼叫者數量和接收方號碼。

格式化後的新值欄位叫call，格式是”Caller number – Receiver number”

完整的程式碼如下所示:

//import util packages

import java.util.HashMap;

import java.util.Map;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

//import Storm IRichBolt package

import backtype.storm.topology.IRichBolt;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Tuple;

//Create a class CallLogCreatorBolt which implement IRichBolt interface

public class CallLogCreatorBolt implements IRichBolt {

   //Create instance for OutputCollector which collects and emits tuples to produce output

   private OutputCollector collector;

   @Override

   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

      this.collector = collector;

   }

   @Override

   public void execute(Tuple tuple) {

      String from = tuple.getString(0);

      String to = tuple.getString(1);

      Integer duration = tuple.getInteger(2);

      collector.emit(new Values(from + " - " + to, duration));

   }

   @Override

   public void cleanup() {}

   @Override

   public void declareOutputFields(OutputFieldsDeclarer declarer) {

      declarer.declare(new Fields("call", "duration"));

   }

   @Override

   public Map<String, Object> getComponentConfiguration() {

      return null;

   }

}

Call log Counter Bolt

Call log counter bolt 接收上一個bolt call及其持續時間作為一個tuple。

在bolt的prepare方法中初始化一個字典(Map)物件。

在execute 方法中,它檢查字典中的tuple併為tuple中的每一個新的“call”值建立一個條目entry，並設定字典的值為1。

對於字典中現有的條目，則將其值+1。

簡而言之,這個bolt在字典中儲存call和它的數量

如果不儲存在字典中，我們也可以把它儲存懂啊一個數據源中。

而不是儲存呼叫及其計數在字典裡,我們也可以將它儲存到一個數據源。

完整的程式程式碼如下：

Coding − CallLogCounterBolt.java

import java.util.HashMap;

import java.util.Map;

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;

import backtype.storm.task.TopologyContext;

import backtype.storm.topology.IRichBolt;

import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.tuple.Tuple;

public class CallLogCounterBolt implements IRichBolt {

   Map<String, Integer> counterMap;

   private OutputCollector collector;

   @Override

   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

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

      this.collector = collector;

   }

   @Override

   public void execute(Tuple tuple) {

      String call = tuple.getString(0);

      Integer duration = tuple.getInteger(1);

      if(!counterMap.containsKey(call)){

         counterMap.put(call, 1);

      }else{

         Integer c = counterMap.get(call) + 1;

         counterMap.put(call, c);

      }

      collector.ack(tuple);

   }

   @Override

   public void cleanup() {

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

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

      }

   }

   @Override

   public void declareOutputFields(OutputFieldsDeclarer declarer) {

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

   }

   @Override

   public Map<String, Object> getComponentConfiguration() {

      return null;

   }

}

Creating Topology

通常storm的topology是一個Thrift結構。

TopologyBuilder類提供了簡單易用的方法來建立複雜的topology。

TopologyBuilder類提供了方法來設spout(setSpout)和bolt(setBolt)。

總之,TopologyBuilder用 createTopology建立topology。

下面的程式碼片段為建立topology的事例：

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt()).shuffleGrouping("call-log-reader-spout");

builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt()).fieldsGrouping("call-log-creator-bolt", new Fields("call"));

shuffleGrouping和fieldsGrouping方法幫助spout和bolt對stream進行分組。

Local Cluster

為了便於開發,我們可以使用“LocalCluster”物件建立一個本地叢集,然後使用“LocalCluster”類的“submitTopology”方法提交topology。

其中，“submitTopology”的引數之一是“Config”類的一個例項。

“Config”類的作用是提交topology之前設定配置選項。

This configuration option will be merged with the cluster configuration at run time and sent to all task (spout and bolt) with the prepare method.

這種配置選項將合併在執行時間和傳送到所有叢集配置任務(壺嘴和螺栓)的準備方法。

一旦topology被提交到叢集,我們需要等待10秒以便叢集計算提交的topology,然後使用”shutdown”方法關閉叢集。

完整的程式程式碼如下：

Coding − LogAnalyserStorm.java

import backtype.storm.tuple.Fields;

import backtype.storm.tuple.Values;

//import storm configuration packages

import backtype.storm.Config;

import backtype.storm.LocalCluster;

import backtype.storm.topology.TopologyBuilder;

//Create main class LogAnalyserStorm submit topology.

public class LogAnalyserStorm {

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

      //Create Config instance for cluster configuration

      Config config = new Config();

      config.setDebug(true);

      //

      TopologyBuilder builder = new TopologyBuilder();

      builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

      builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())

         .shuffleGrouping("call-log-reader-spout");

      builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())

         .fieldsGrouping("call-log-creator-bolt", new Fields("call"));

      LocalCluster cluster = new LocalCluster();

      cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology());

      Thread.sleep(10000);

      //Stop the topology

      cluster.shutdown();

   }

}

Building and Running the Application

完整的應用程式有四個Java程式碼：

1.FakeCallLogReaderSpout.java2.CallLogCreaterBolt.java3.CallLogCounterBolt.java4.LogAnalyerStorm.java

應用程式可以使用下面的命令構建：

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*” *.java

應用程式可以使用以下的命令執行：

java -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:.LogAnalyserStorm

Output

一旦應用程式啟動,它將輸出完整的叢集啟動程序的細節,和pout和bolt處理過程,最後,叢集關閉這些處理過程。

在 "CallLogCounterBolt"程式碼中,我們列印了call和count的具體資訊。

這些資訊將顯示在控制檯，如下：

1234123402 - 1234123401 : 781234123402 - 1234123404 : 881234123402 - 1234123403 : 1051234123401 - 1234123404 : 741234123401 - 1234123403 : 811234123401 - 1234123402 : 811234123403 - 1234123404 : 861234123404 - 1234123401 : 631234123404 - 1234123402 : 821234123403 - 1234123402 : 831234123404 - 1234123403 : 861234123403 - 1234123401 : 93

JVM外的其他語言

Storm topology通過Thrift介面實現，這使得很容易任何語言去提交topology到storm叢集中。

Storm支援Ruby、Python和許多其他語言。

讓我們看看使用python事例：

Python Binding

Python是一種解釋型,互動的、面向物件的高階程式語言。

Storm支援用Python實現其topology。

Python支援 emitting, anchoring, acking, and logging operations。

如你所知,bolt可以使用任何語言定義。

Bolts written in another language are executed as sub-processes, and Storm communicates with those sub-processes with JSON messages over stdin/stdout.

下面來看一個用python編寫的bolt來計算單詞出現次數的事例：

public static class WordCount implements IRichBolt {public WordSplit() {super("python", "splitword.py");}public void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields("word"));}}

Here the class WordCount implements the IRichBolt interface and running with python implementation specified super method argument "splitword.py".

現在建立一個名為“splitword.py”的python實現。

import stormclass WordCountBolt(storm.BasicBolt):def process(self, tup):words = tup.values[0].split(" ")for word in words:storm.emit([word])WordCountBolt().run()

這是Python實現計數的示例。

同樣你也可以用其他支援的語言實現。