Input and output does not present in Storm UI.In Storm UI you can see no of tuples emitted,processing time,cluster configuration and health of your cluster.To see output and input use logger mechanism and then check each worker log files which present in log folder of storm package.
To create a topology in Storm you need two things one spout and one bolt.Please find below sample coded :-
SampleSpout.java
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
public class SampleSpout implements IRichSpout{
SpoutOutputCollector collector;
int i=0;
List<Object> tupleList;
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
// TODO Auto-generated method stub
}
@Override
public void close() {
// TODO Auto-generated method stub
}
@Override
public void activate() {
// TODO Auto-generated method stub
}
@Override
public void deactivate() {
// TODO Auto-generated method stub
}
@Override
public void nextTuple() {
tupleList=new ArrayList<Object>();
tupleList.add("storm"+i);
tupleList.add(i);
collector.emit(tupleList,i);
i++;
}
@Override
public void ack(Object msgId) {
// TODO Auto-generated method stub
}
@Override
public void fail(Object msgId) {
// TODO Auto-generated method stub
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word","count"));
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
SampleBolt.java
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
public class SampleBolt implements IBasicBolt {
private static Logger log = LoggerFactory.getLogger(SampleBolt.class);
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
@Override
public void prepare(Map stormConf, TopologyContext context) {
// TODO Auto-generated method stub
}
@Override
public void execute(Tuple input, BasicOutputCollector collector) {
log.info(input.getValues().toString()+"output values");
}
@Override
public void cleanup() {
// TODO Auto-generated method stub
}
}
SampleTopology.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
public class SampleTopology {
/**
* @param args
*/
public static void main(String[] args) {
TopologyBuilder topology=new TopologyBuilder();
topology.setSpout("sampleSpout",new SampleSpout());
topology.setBolt("sampleBolt",new SampleBolt()).shuffleGrouping("sampleSpout");
Config conf = new Config();
conf.setDebug(true);
LocalCluster cluster=new LocalCluster();
cluster.submitTopology("test", conf, topology.createTopology());
}
}
与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…