public static void main(String[] args) throws IOException{


Configuration conf1 = new Configuration();
//第一个job的配置
Job job1 = new Job(conf1);
job1.setJarByClass(LiFangMinHash.class);
job1.setNumReduceTasks(25);
job1.setMapperClass(maphadoop1.class);
job1.setReducerClass(reducehadoop1.class);
job1.setOutputFormatClass(TextOutputFormat.class);
job1.setOutputKeyClass(Text.class);
job1.setOutputValueClass(Text.class);
ControlledJob ctrljob1=new ControlledJob(conf1);
ctrljob1.setJob(job1);
FileInputFormat.addInputPath(job1, new Path(args[0]));
FileOutputFormat.setOutputPath(job1, new Path(args[1]));

Configuration conf2 = new Configuration();
Job job2=new Job(conf2);
job2.setJarByClass(LiFangMinHash.class);
job2.setNumReduceTasks(25);
job2.setMapperClass(maphadoop2.class);
job2.setReducerClass(reducehadoop2.class);
job2.setOutputFormatClass(TextOutputFormat.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(Text.class);
ControlledJob ctrljob2=new ControlledJob(conf2);
ctrljob2.setJob(job2);

ctrljob2.addDependingJob(ctrljob1);

FileInputFormat.addInputPath(job2, new Path(args[1]));
FileOutputFormat.setOutputPath(job2,new Path(args[2]));



JobControl jobCtrl=new JobControl("myctrl");
jobCtrl.addJob(ctrljob1);
jobCtrl.addJob(ctrljob2);
jobCtrl.run();

while(true){


if(jobCtrl.allFinished()){//如果作业成功完成，就打印成功作业的信息
System.out.println(jobCtrl.getSuccessfulJobList());
jobCtrl.stop();
break;
}
}
}
}