Apache Storm in Yahoo! Finanza
Yahoo! Finance è il sito Web di notizie economiche e dati finanziari leader in Internet. Fa parte di Yahoo! e fornisce informazioni su notizie finanziarie, statistiche di mercato, dati di mercato internazionale e altre informazioni sulle risorse finanziarie a cui chiunque può accedere.
Se sei un utente Yahoo! registrato utente, puoi personalizzare Yahoo! Finanza per trarre vantaggio dalle sue determinate offerte. Yahoo! L'API Finance viene utilizzata per eseguire query sui dati finanziari da Yahoo!
Questa API visualizza i dati in ritardo di 15 minuti rispetto al tempo reale e aggiorna il database ogni minuto per accedere alle informazioni correnti relative alle scorte. Ora prendiamo uno scenario in tempo reale di un'azienda e vediamo come generare un avviso quando il valore delle sue azioni scende al di sotto di 100.
Creazione beccuccio
Lo scopo del beccuccio è quello di ottenere i dettagli dell'azienda ed emettere i prezzi a bulloni. È possibile utilizzare il seguente codice di programma per creare uno spout.
Codifica: 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;
}
}
Creazione di bulloni
Qui lo scopo del bullone è elaborare i prezzi della società data quando i prezzi scendono al di sotto di 100. Utilizza l'oggetto Java Map per impostare l'avviso del limite di prezzo di interruzione come truequando i prezzi delle azioni scendono al di sotto di 100; altrimenti falso. Il codice completo del programma è il seguente:
Codifica: 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;
}
}
Invio di una topologia
Questa è l'applicazione principale in cui YahooFinanceSpout.java e PriceCutOffBolt.java sono collegati insieme e producono una topologia. Il codice di programma seguente mostra come inviare una topologia.
Codifica: 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();
}
}
Creazione ed esecuzione dell'applicazione
L'applicazione completa ha tre codici Java. Sono i seguenti:
- YahooFinanceSpout.java
- PriceCutOffBolt.java
- YahooFinanceStorm.java
L'applicazione può essere creata utilizzando il seguente comando:
javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/yahoofinance/lib/*” *.java
L'applicazione può essere eseguita utilizzando il seguente comando:
javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/yahoofinance/lib/*”:.
YahooFinanceStorm
Produzione
L'output sarà simile al seguente:
GOOGL : false
AAPL : false
INTC : true