storm任务示例
?LogProcess.java
package mytest;
?
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.Map;
?
import mytest.ThroughputTest.GenSpout;
?
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
?
public class LogProcess {
?????????public static class FileSpout extends BaseRichSpout {
?
???????????????????/**
????????????????????*/
???????????????????private static final long serialVersionUID = 1L;
???????????????????private SpoutOutputCollector _collector;
???????????????????private BufferedReader br;
???????????????????private String dataFile;
??????????????????
???????????????????//定义spout文件
???????????????????FileSpout(String dataFile){
????????????????????????????this.dataFile = dataFile;
???????????????????}
?
???????????????????//定义如何读取spout文件
???????????????????@Override
???????????????????public void open(Map conf, TopologyContext context,
?????????????????????????????????????SpoutOutputCollector collector) {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????_collector = collector;
????????????????????????????File csv = new File(dataFile); // log file
????????????????????????????try {
?????????????????????????????????????br = new BufferedReader(new FileReader(csv));
????????????????????????????} catch (FileNotFoundException e) {
?????????????????????????????????????// TODO Auto-generated catch block
?????????????????????????????????????e.printStackTrace();
????????????????????????????}
???????????????????}
?
???????????????????//获取下一个tuple的方法
???????????????????@Override
???????????????????public void nextTuple() {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????try {
????????????????????????????????????
?????????????????????????????????????String line = null;
?????????????????????????????????????while ((line = br.readLine()) != null) {
???????????????????????????????????????????????_collector.emit(new Values(line));
?????????????????????????????????????}
????????????????????????????} catch (FileNotFoundException e) {
?????????????????????????????????????// TODO Auto-generated catch block
?????????????????????????????????????e.printStackTrace();
????????????????????????????} catch (IOException e) {
?????????????????????????????????????// TODO Auto-generated catch block
?????????????????????????????????????e.printStackTrace();
????????????????????????????}
???????????????????}
?
?
???????????????????@Override
???????????????????public void declareOutputFields(OutputFieldsDeclarer declarer) {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????declarer.declare(new Fields("line"));
???????????????????}
??????????????????
?????????}
????????
?
?????????public static class Process extends BaseRichBolt{
?
???????????????????private String _seperator;
???????????????????private String _outFile;
???????????????????PrintWriter pw;
???????????????????private OutputCollector _collector;
???????????????????private BufferedWriter bw;
??????????????????
???????????????????public Process(String seperator,String outFile) {
????????????????????????????this._seperator = seperator;
????????????????????????????this._outFile???= outFile;
???????????????????????????
???????????????????}
??????????????????
???????????????????//把输出结果保存到外部文件里面。
???????????????????@Override
???????????????????public void prepare(Map stormConf, TopologyContext context,
?????????????????????????????????????OutputCollector collector) {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????this._collector = collector;
????????????????????????????File out = new File(_outFile);
????????????????????????????try {
//??????????????????????????????????br = new BufferedWriter(new FileWriter(out));
?????????????????????????????????????bw = new BufferedWriter(new OutputStreamWriter(?
?????????????????????????????new FileOutputStream(out, true)));?
????????????????????????????} catch (IOException e1) {
?????????????????????????????????????// TODO Auto-generated catch block
?????????????????????????????????????e1.printStackTrace();
????????????????????????????}????????????????
???????????????????}
??????????????????
???????????????????//blot计算单元，把tuple中的数据添加一个bkeep和回车。然后保存到outfile指定的文件中。
???????????????????@Override
???????????????????public void execute(Tuple input) {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????String line = input.getString(0);
//?????????????????????????System.out.println(line);
????????????????????????????String[] str = line.split(_seperator);
????????????????????????????System.out.println(str[2]);
????????????????????????????try {
?????????????????????????????????????bw.write(str[2]+",bkeep"+"\n");
?????????????????????????????????????bw.flush();
????????????????????????????} catch (IOException e) {
?????????????????????????????????????// TODO Auto-generated catch block
?????????????????????????????????????e.printStackTrace();
????????????????????????????}
???????????????????????????
????????????????????????????_collector.emit(new Values(line));
???????????????????}
?
???????????????????@Override
???????????????????public void declareOutputFields(OutputFieldsDeclarer declarer) {
????????????????????????????// TODO Auto-generated method stub
????????????????????????????declarer.declare(new Fields("line"));
???????????????????}
??????????????????
?????????}
????????
?????????public static void main(String[] argv) throws AlreadyAliveException, InvalidTopologyException{
???????????????????String dataFile = argv[0];?//输入文件
???????????????????String seperator = argv[1];??????//分隔符
???????????????????String outFile???= argv[2];?//输出文件
???????????????????boolean distribute = Boolean.valueOf(argv[3]);???????//本地模式还是集群模式
???????????????????TopologyBuilder builder = new TopologyBuilder();??//build一个topology
????????builder.setSpout("spout", new FileSpout(dataFile), 1);???//指定spout
????????builder.setBolt("bolt", new Process(seperator,outFile),1).shuffleGrouping("spout");??//指定bolt，包括bolt、process和grouping
????????Config conf = new Config();
????????if(distribute){
????????????StormSubmitter.submitTopology("LogProcess", conf, builder.createTopology());
????????}else{
?????????????????LocalCluster cluster = new LocalCluster();
?????????????????cluster.submitTopology("LogProcess", conf, builder.createTopology());
????????}
?????????}???????
}
?
?
运行
[admin@vkvm161064 guandao]$ pwd
/home/admin/guandao
[admin@vkvm161064 guandao]$ ls
out.txt??storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jar??tmp.txt
?
输入文件：
[admin@vkvm161064 guandao]$ cat tmp.txt
a,b,c,d
1,2,3,4
A,B,C,D
xx,ff,ff,ss
xx,ff,alibaba,ss
xx,ff,taobao,ss
xx,xx,xx,xx
xx,xx,ll,xx
xx,xx,hero,xx
?
输出文件：
[admin@vkvm161064 guandao]$ cat out.txt
c,bkeep
3,bkeep
C,bkeep
ff,bkeep
alibaba,bkeep
taobao,bkeep
xx,bkeep
ll,bkeep
hero,bkeep
?
提交topology
[admin@vkvm161064 guandao]$ storm jar ./storm-starter-0.0.1-SNAPSHOT-jar-with-dependencies.jarSHOT-jar-with-dependencies.jar mytest.LogProcess /home/admin/guandao/tmp.txt , /home/admin/guandao/out.txt fase
?
语法：storm??jar?自己开发的topology???topology_name??inputfile?分隔符???outputfile??true/false(true代表集群运行)
