Storm框架原本是設計用來做網際網路短文字處理和一些統計工作的，是一種分散式流式計算框架。在一些場合，特別是在已經用了Storm架構以後，發現又想用EPL語句，Storm和類SQL語言EPL結合也不失為一種方案。對線上規則修改、視窗資料可靠性要求高的地方還有用專用的CEP叢集方案、Hot-Hot HA等方案可供選擇，不過這些不是本文的重點。本文的重點就是讓Storm插上EPL的翅膀，輕鬆解決一些實際專案中的攔路虎。

1.示例操作步驟

前3步是安裝Storm（或jStorm），已經安裝了Storm（或jStorm）的讀者可以跳過，直接到第4步。

（1）下載安裝jdk

（2）下載安裝 zookeeper,啟動zookeeper

     exportZOOKEEPER_HOME="/path/to/zookeeper"
     exportPATH=$PATH:$ZOOKEEPER_HOME/bin
     cp/usr/local/zookeeper/conf/zoo_sample.cfg /usr/local/zookeeper/conf/zoo.cfg  

     zkServer.sh start

（3）下載storm,

解壓到linux伺服器相應位置，也可以下載其它版本。

（4）EPL引擎配置目錄，

cep_home目錄拷貝到storm的目錄中，如apache-storm-0.9.4/cep_home。這個目錄也可以自定義，需修改EPLExampleTopology.java程式碼中的cep_home變數的值。

lib/sodbasecep/目錄下jar檔案拷貝到storm的lib目錄下

dist目錄下的cep-application-storm-example.jar拷貝到storm的examples目錄下

注：使用jStorm的讀者，配置方法和Storm一致，本系列文章中在Storm上執行EPL的示例可以同時在Storm和jStorm上執行。

（5）啟動storm

# bin/storm nimbus &

# bin/stormsupervisor &

（6）執行Topology

執行命令啟動topology

 bin/storm jar examples/cep-application-storm-example.jar com.sodbase.integration.storm.EPLExampleTopology
 

（7）輸出結果（本示例的功能是IT系統監控中的分析函式呼叫響應時間，是否超時）

2. 工作原理

Storm中使用SQL,較為簡便的方法就是使用EPLBolt。這種型別的一個Bolt就代表著一個SQL(EPL)語句，有時也能代表多個SQL語句。

2.1 EPL建模

首先，建議大家用Studio建模，建模後匯出為XML檔案，能大大提高建模速度。本例的EPL XML檔案在files目錄下的calltimeout.xml

EPL的輸入流名稱使用com.sodbase.inputadaptor.StubInputAdaptor進行宣告，從而在EPL中可以使用。

EPL的輸出流使用com.sodbase.outputadaptor.storm.SodbaseCepStormOutputAdaptor介面卡

    <outputAdaptors>
        <isOutputAsSelection>true</isOutputAsSelection>
        <outputAdaptorClassName>com.sodbase.outputadaptor.storm.SodbaseCepStormOutputAdaptor</outputAdaptorClassName>
        <isExternal>false</isExternal>
        <queryName>calltimeoutnotification</queryName>
    </outputAdaptors>

查詢響應時間的語句

CREATE QUERY callnottimeout 
SELECT T2._start_time_-T1._start_time_ AS responsetime, T1.callerid AS functionname ,'false' AS timeout 
FROM T1:callstream,T2:callstream 
PATTERN T1&T2 
WHERE T1.callerid=T2.callerid  AND T1.eventtype ='start' AND T2.eventtype ='end' 
WITHIN 1000 

CREATE QUERY calltimeoutnotification 
SELECT '-1' AS responsetime, T1.callerid AS functionname,'true' AS timeout 
FROM T1:callstream,T2:callstream,T3:calltimeoutevent 
PATTERN T1;!T2;T3  
WHERE T3._end_time_-T1._end_time_=1000 AND T2.callerid=T3.callerid AND T1.eventtype='start' 
WITHIN 1000 

CREATE QUERY calltimeout 
SELECT 'timer' AS type,T1.time AS time,T1.callerid  AS callerid 
FROM T1:callstream 
PATTERN T1 WHERE T1.eventtype ='start' 
WITHIN 0

<outputAdaptors>
    <isOutputAsSelection>true</isOutputAsSelection>
    <outputAdaptorClassName>com.sodbase.outputadaptor.timer.FixedDelayTimerOutputAdaptor</outputAdaptorClassName>
    <adaptorParams>calltimeoutevent</adaptorParams>
    <adaptorParams>1000</adaptorParams>
    <isExternal>false</isExternal>
    <queryName>calltimeout</queryName>
</outputAdaptors>

2.在storm中使用EPL模型

（1）將EPL需要的包拷貝到storm的lib目錄下，再啟動nimbus和supervisor

sodbase-cep-engine.jar  sodbase-dataadaptor-storm.jar  sodbase-dataadaptor-timer.jar  sodbase-studio.jar  xalan-2.7.1.jar  xercesImpl-2.9.1.jar  xml-apis-1.3.04.jar

（2）在storm目錄下建一個目錄cep_home,存放EPL引擎的配置檔案和日誌。把configuration資料夾、logging.properties放到cep_home下面。cep_home的路徑可以自定義，和下面EPLExampleTopology 程式碼中一致即可。

（3）EPL XML模型檔案放到放到cep_home/files目錄下，位置也可自定義，要和下面EPLExampleTopology 程式碼中一致。

（4）編寫Topology應用EPLExampleTopology， 程式碼如下

public class EPLExampleTopology {

  public static void main(String[] args) throws Exception {
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("event", new RandomEventSpout(), 1);
    //TODO: to specify stream name that EPL uses, and the output fields
    Fields outputFields = new Fields("functionname","responsetime","timeout");
    String cep_home="/user/apache-storm-0.9.4/cep_home";
    String cepmodelfile="/user/apache-storm-0.9.4/cep_home/files/calltimeout.xml";
    builder.setBolt("EPL", new EPLBolt("callstream",outputFields,cep_home,cepmodelfile), 1).shuffleGrouping("event");
    builder.setBolt("print1", new PrintBolt(), 1).shuffleGrouping("EPL");

    Config conf = new Config();
    conf.setDebug(false);

    if (args != null && args.length > 0) {
      conf.setNumWorkers(3);

      StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
    }
    else {

      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("test", conf, builder.createTopology());
      Utils.sleep(10000);
      cluster.killTopology("test");
      cluster.shutdown();
    }
  }
}

builder.setBolt("EPL", new EPLBolt("callstream",outputFields,cep_home,cepmodelfile), 1).shuffleGrouping("event");

表示將"event" Spout的輸出，接到EPL引擎的streamname "callstream"，輸出outputFieds。用了EPLBolt專門處理EPL語句，建構函式有4個引數

streamname:與EPL模型中StubInputAdaptor的名稱對應

outputFields:Bolt的輸出欄位，應與EPL模型中SELECT語句一致

cep_home:EPL引擎配置檔案和日誌檔案目錄

cepmodelfile:EPL XML模型檔案的路徑

（5）EPLExampleTopology的RandomEventSpout和PrintBolt

RandomEventSpout提供呼叫開始結束事件模擬資料的spout

public class RandomEventSpout extends BaseRichSpout {
  SpoutOutputCollector _collector;
  Random _rand;
  @Override
  public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    _collector = collector;
    _rand = new Random();
  }

  @Override
  public void nextTuple() {
    
    int functionid=_rand.nextInt(10000);
    String a = "function-"+functionid;
    String b = "start";    
    _collector.emit(new Values(a,b));
    Utils.sleep(500);
    if(functionid%2==0)
    {    	
        b = "end";    
        _collector.emit(new Values(a,b));
    }
  }

  @Override
  public void ack(Object id) {
  }

  @Override
  public void fail(Object id) {
  }

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

}

public  class PrintBolt extends BaseRichBolt {
    OutputCollector _collector;  
    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
    }
    public void execute(Tuple tuple) {
    	Fields fields = tuple.getFields();
    	for(String field:fields)
    		System.out.print(field+": "+tuple.getValueByField(field));
    	System.out.println();	      
    	_collector.ack(tuple);
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
  }

（6）將Topology打包成jar,在storm中測試執行。

本文示例的功能是監控系統中監測服務呼叫或系統呼叫的響應時間，是否超時。

參考：

SODBASE CEP用於輕鬆、高效實施資料監測、監控類、交易類、日誌分析類專案。EPL語法見SODSQL寫法與示例。圖形化建模請使用SODBASE Studio。