Hadoop基本流程与应用开发

huiselele

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 = 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);//构建Config
　　FileSystem fileSys = FileSystem.get(conf);
　　fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));//将本地文件系统的文件拷贝到HDFS中
　　conf.setJobName("analysisjob");
　　conf.setOutputKeyClass(Text.class);//输出的key类型，在OutputFormat会检查
　　conf.setOutputValueClass(LongWritable.class); //输出的value类型，在OutputFormat会检查
　　conf.setMapperClass(MapClass.class);
　　conf.setCombinerClass(CombinerClass.class);
　　conf.setReducerClass(ReduceClass.class);
　　conf.setPartitionerClass(PartitionerClass.class);
　　conf.set("mapred.reduce.tasks", "2");//强制需要有两个Reduce来分别处理流量和次数的统计
　　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.");
　　//删除输入和输出的临时文件
　　fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath));
　　fileSys.delete(new Path(shortin),true);
　　fileSys.delete(new Path(shortout),true);
　　}