public static void main(String[] args) throws Exception, IOException, InterruptedException {
  Configuration conf = new Configuration();
  Job job = Job.getInstance(conf);
  FileSystem fs=FileSystem.get(conf);
  job.setJarByClass(FlowBeanRunner.class);
 
  job.setMapperClass(FlowBeanMapper.class);
  job.setReducerClass(FlowBeanReduce.class);
 
  job.setOutputKeyClass(FlowBean.class);
  job.setOutputValueClass(Text.class);
 
  //指定自定義的partitioner類，替換掉框架預設的HashPartitioner
  job.setPartitionerClass(ProvincePartitioner.class);
  //指定reduce task數量，跟ProvincePartitioner的分割槽數匹配
  job.setNumReduceTasks(6);     //Reduce的個數在這裡設定  
 
  //要處理的資料所在的path
  //指定資料夾即可，該資料夾下的所有檔案都會被處理
  FileInputFormat.setInputPaths(job, new Path("/phone/logs"));
 
  Path destPath= new Path("/phone/cleanlogs");
  if(fs.exists(destPath)){
   fs.delete(destPath, true);
  }
 
  //處理完得到的結果輸出的path
  FileOutputFormat.setOutputPath(job, destPath);//提交本次作業
  job.waitForCompletion(true);
 }