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