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代表集群运行)