@Override
	public int run(String[] args) throws Exception {
		//String fileName = "E:/tmp/1111.txt";
		Configuration conf = new Configuration();
		conf.set("fs.default.name", "file:///");
		conf.set("mapred.job.tracker", "local");
		String inputFile = "E:/tmp/call.pl";
		String output = "E:/tmp/output/";
		
		Job job = new Job(conf,this.getClass().getSimpleName());
		job.setJarByClass(this.getClass());
		job.setMapperClass(LxwMapper.class);
		job.setReducerClass(LxwReducer.class);
		
		FileInputFormat.addInputPath(job, new Path(inputFile));
		FileOutputFormat.setOutputPath(job, new Path(output));
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new Configuration(), new LxwJob(), args);
		System.exit(exitCode);
	}