Apache Storm in Yahoo! Finanzen

Yahoo! Finance ist die führende Website für Wirtschaftsnachrichten und Finanzdaten im Internet. Es ist ein Teil von Yahoo! und gibt Informationen über Finanznachrichten, Marktstatistiken, internationale Marktdaten und andere Informationen über Finanzressourcen, auf die jeder zugreifen kann.

Wenn Sie ein registrierter Yahoo! Benutzer, dann können Sie Yahoo! Finanzen, um von seinen bestimmten Angeboten zu profitieren. Yahoo! Die Finanz-API wird verwendet, um Finanzdaten von Yahoo!

Diese API zeigt Daten an, die gegenüber der Echtzeit um 15 Minuten verzögert sind, und aktualisiert ihre Datenbank alle 1 Minute, um auf aktuelle aktienbezogene Informationen zuzugreifen. Nehmen wir nun ein Echtzeitszenario eines Unternehmens und sehen, wie eine Warnung ausgelöst wird, wenn der Aktienwert unter 100 fällt.

Auslauferstellung

Der Zweck des Auslaufs ist es, die Details des Unternehmens zu erhalten und die Preise an Schrauben abzugeben. Mit dem folgenden Programmcode können Sie einen Auslauf erstellen.

Codierung: 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;
   }
	
}

Bolzenerstellung

Hier besteht der Zweck von Bolt darin, die Preise des angegebenen Unternehmens zu verarbeiten, wenn die Preise unter 100 fallen. Es verwendet das Java Map-Objekt, um die Warnung für das Grenzpreislimit als festzulegen truewenn die Aktienkurse unter 100 fallen; sonst falsch. Der vollständige Programmcode lautet wie folgt:

Codierung: 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;
   }
	
}

Einreichen einer Topologie

Dies ist die Hauptanwendung, in der YahooFinanceSpout.java und PriceCutOffBolt.java miteinander verbunden sind und eine Topologie erstellen. Der folgende Programmcode zeigt, wie Sie eine Topologie senden können.

Codierung: 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();
   }
}

Erstellen und Ausführen der Anwendung

Die vollständige Anwendung verfügt über drei Java-Codes. Sie sind wie folgt -

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

Die Anwendung kann mit dem folgenden Befehl erstellt werden:

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

Die Anwendung kann mit dem folgenden Befehl ausgeführt werden:

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

Ausgabe

Die Ausgabe ist ähnlich wie folgt:

GOOGL : false
AAPL : false
INTC : true