ivaneeo's blog

    public static class MapClass extends MapReduceBase

        implements Mapper<LongWritable, Text, Text, LongWritable> 

    {

        public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter)

                throws IOException

        {    

            String line = value.toString();//沒有配置RecordReader，所以默認(rèn)采用line的實(shí)現(xiàn)，key就是行號(hào)，value就是行內(nèi)容

            if (line == null || line.equals(""))

                return;

            String[] words = line.split(",");

            if (words == null || words.length < 8)

                return;

            String appid = words[1];

            String apiName = words[2];

            LongWritable recbytes = new LongWritable(Long.parseLong(words[7]));

            Text record = new Text();

            record.set(new StringBuffer("flow::").append(appid)

                            .append("::").append(apiName).toString());

            reporter.progress();

            output.collect(record, recbytes);//輸出流量的統(tǒng)計(jì)結(jié)果，通過flow::作為前綴來標(biāo)示。

            record.clear();

            record.set(new StringBuffer("count::").append(appid).append("::").append(apiName).toString());

            output.collect(record, new LongWritable(1));//輸出次數(shù)的統(tǒng)計(jì)結(jié)果，通過count::作為前綴來標(biāo)示

        }    

    }

    public static class PartitionerClass implements Partitioner<Text, LongWritable>

    {

        public int getPartition(Text key, LongWritable value, int numPartitions)

        {

            if (numPartitions >= 2)//Reduce 個(gè)數(shù)，判斷流量還是次數(shù)的統(tǒng)計(jì)分配到不同的Reduce

                if (key.toString().startsWith("flow::"))

                    return 0;

                else

                    return 1;

            else

                return 0;

        }

        public void configure(JobConf job){}    

}

    public static class ReduceClass extends MapReduceBase

        implements Reducer<Text, LongWritable,Text, LongWritable> 

    {

        public void reduce(Text key, Iterator<LongWritable> values,

                OutputCollector<Text, LongWritable> output, Reporter reporter)throws IOException

        {

            Text newkey = new Text();

            newkey.set(key.toString().substring(key.toString().indexOf("::")+2));

            LongWritable result = new LongWritable();

            long tmp = 0;

            int counter = 0;

            while(values.hasNext())//累加同一個(gè)key的統(tǒng)計(jì)結(jié)果

            {

                tmp = tmp + values.next().get();

                

                counter = counter +1;//擔(dān)心處理太久，JobTracker長時(shí)間沒有收到報(bào)告會(huì)認(rèn)為TaskTracker已經(jīng)失效，因此定時(shí)報(bào)告一下

                if (counter == 1000)

                {

                    counter = 0;

                    reporter.progress();

                }

            }

            result.set(tmp);

            output.collect(newkey, result);//輸出最后的匯總結(jié)果

        }    

    }

	public static void main(String[] args)

{

try

{

run(args);

} catch (Exception e)

{

e.printStackTrace();

}

}

public static void run(String[] args) throws Exception

{

if (args == null || args.length <2)

{

System.out.println("need inputpath and outputpath");

return;

}

String inputpath = args[0];

String outputpath = args[1];

String shortin = args[0];

String shortout = args[1];

if (shortin.indexOf(File.separator) >= 0)

shortin = shortin.substring(shortin.lastIndexOf(File.separator));

if (shortout.indexOf(File.separator) >= 0)

shortout = shortout.substring(shortout.lastIndexOf(File.separator));

SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd");

shortout = new StringBuffer(shortout).append("-")

.append(formater.format(new Date())).toString();





if (!shortin.startsWith("/"))

shortin = "/" + shortin;

if (!shortout.startsWith("/"))

shortout = "/" + shortout;

shortin = "/user/root" + shortin;

shortout = "/user/root" + shortout;			

File inputdir = new File(inputpath);

File outputdir = new File(outputpath);

if (!inputdir.exists() || !inputdir.isDirectory())

{

System.out.println("inputpath not exist or isn't dir!");

return;

}

if (!outputdir.exists())

{

new File(outputpath).mkdirs();

}



JobConf conf = new JobConf(new Configuration(),LogAnalysiser.class);//構(gòu)建Config

FileSystem fileSys = FileSystem.get(conf);

fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));//將本地文件系統(tǒng)的文件拷貝到HDFS中



conf.setJobName("analysisjob");

conf.setOutputKeyClass(Text.class);//輸出的key類型，在OutputFormat會(huì)檢查

conf.setOutputValueClass(LongWritable.class); //輸出的value類型，在OutputFormat會(huì)檢查

conf.setMapperClass(MapClass.class);

conf.setCombinerClass(CombinerClass.class);

conf.setReducerClass(ReduceClass.class);

conf.setPartitionerClass(PartitionerClass.class);

conf.set("mapred.reduce.tasks", "2");//強(qiáng)制需要有兩個(gè)Reduce來分別處理流量和次數(shù)的統(tǒng)計(jì)

FileInputFormat.setInputPaths(conf, shortin);//hdfs中的輸入路徑

FileOutputFormat.setOutputPath(conf, new Path(shortout));//hdfs中輸出路徑



Date startTime = new Date();

System.out.println("Job started: " + startTime);

JobClient.runJob(conf);

Date end_time = new Date();

System.out.println("Job ended: " + end_time);

System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");

//刪除輸入和輸出的臨時(shí)文件

fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath));

fileSys.delete(new Path(shortin),true);

fileSys.delete(new Path(shortout),true);

}

public class ExampleDriver {

  public static void main(String argv[]){

    ProgramDriver pgd = new ProgramDriver();

    try {

      pgd.addClass("analysislog", LogAnalysiser.class, "A map/reduce program that analysis log .");

      pgd.driver(argv);

    }

    catch(Throwable e){

      e.printStackTrace();

    }

  }

}

hadoop jar analysiser.jar analysislog /home/wenchu/test-in /home/wenchu/test-out