Apache Storm在雅虎財(cái)經(jīng)上的應(yīng)用

雅虎財(cái)經(jīng)是互聯(lián)網(wǎng)領(lǐng)先的商業(yè)新聞和金融數(shù)據(jù)網(wǎng)站。它是雅虎的一部分，并提供有關(guān)金融新聞，市場(chǎng)統(tǒng)計(jì)，國(guó)際市場(chǎng)數(shù)據(jù)和其他任何人都可以訪問的財(cái)務(wù)資源信息。

如果您是注冊(cè)的Yahoo!用戶，那么您可以自定義Yahoo! Finance以利用其特定產(chǎn)品。Yahoo! Finance API用于從Yahoo!查詢財(cái)務(wù)數(shù)據(jù)。

此API顯示實(shí)時(shí)延遲15分鐘的數(shù)據(jù)，并每1分鐘更新其數(shù)據(jù)庫，以訪問當(dāng)前股票相關(guān)信息。現(xiàn)在讓我們看一家公司的實(shí)時(shí)情景，看看當(dāng)公司的股票價(jià)值低于100時(shí)如何提高警報(bào)。

Spout創(chuàng)建

spout的目的是獲得公司的詳細(xì)信息，并發(fā)出價(jià)格spout。您可以使用以下程序代碼創(chuàng)建spout。

編碼：YahooFinanceSpout.java

import java.util.*;
import java.io.*;
import java.math.BigDecimal;

//import yahoofinace packages
import yahoofinance.YahooFinance;
import yahoofinance.Stock;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;

public class YahooFinanceSpout implements IRichSpout {
   private SpoutOutputCollector collector;
   private boolean completed = false;
   private TopologyContext context;
	
   @Override
   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector){
      this.context = context;
      this.collector = collector;
   }

   @Override
   public void nextTuple() {
      try {
         Stock stock = YahooFinance.get("INTC");
         BigDecimal price = stock.getQuote().getPrice();

         this.collector.emit(new Values("INTC", price.doubleValue()));
         stock = YahooFinance.get("GOOGL");
         price = stock.getQuote().getPrice();

         this.collector.emit(new Values("GOOGL", price.doubleValue()));
         stock = YahooFinance.get("AAPL");
         price = stock.getQuote().getPrice();

         this.collector.emit(new Values("AAPL", price.doubleValue()));
      } catch(Exception e) {}
   }

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

   @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創(chuàng)建

這里的的目的是當(dāng)價(jià)格低于100時(shí)處理給定公司的價(jià)格。它使用Java Map對(duì)象在股價(jià)低于100時(shí)設(shè)置截止價(jià)格限制警報(bào)為真;否則為false。完整的程序代碼如下 -

編碼：PriceCutOffBolt.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 PriceCutOffBolt implements IRichBolt {
   Map<String, Integer> cutOffMap;
   Map<String, Boolean> resultMap;
	
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.cutOffMap = new HashMap <String, Integer>();
      this.cutOffMap.put("INTC", 100);
      this.cutOffMap.put("AAPL", 100);
      this.cutOffMap.put("GOOGL", 100);

      this.resultMap = new HashMap<String, Boolean>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String company = tuple.getString(0);
      Double price = tuple.getDouble(1);

      if(this.cutOffMap.containsKey(company)){
         Integer cutOffPrice = this.cutOffMap.get(company);

         if(price < cutOffPrice) {
            this.resultMap.put(company, true);
         } else {
            this.resultMap.put(company, false);
         }
      }
		
      collector.ack(tuple);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Boolean> entry:resultMap.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("cut_off_price"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

提交拓?fù)?/h2>

這是YahooFinanceSpout.java和PriceCutOffBolt.java連接在一起并生成拓?fù)涞闹饕獞?yīng)用程序。以下程序代碼顯示了如何提交拓?fù)洹?/span>

編碼：YahooFinanceStorm.java

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

public class YahooFinanceStorm {
   public static void main(String[] args) throws Exception{
      Config config = new Config();
      config.setDebug(true);
		
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("yahoo-finance-spout", new YahooFinanceSpout());

      builder.setBolt("price-cutoff-bolt", new PriceCutOffBolt())
         .fieldsGrouping("yahoo-finance-spout", new Fields("company"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("YahooFinanceStorm", config, builder.createTopology());
      Thread.sleep(10000);
      cluster.shutdown();
   }
}

構(gòu)建和運(yùn)行應(yīng)用程序

完整的應(yīng)用程序有三個(gè)Java代碼。他們?nèi)缦?-

• YahooFinanceSpout.java
• PriceCutOffBolt.java
• YahooFinanceStorm.java

應(yīng)用程序可以使用以下命令構(gòu)建 -

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

應(yīng)用程序可以使用以下命令運(yùn)行 -

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/yahoofinance/lib/*”:.
YahooFinanceStorm

輸出

輸出將類似于以下內(nèi)容 -

GOOGL : false
AAPL : false
INTC : true