How to implement serial execution of mapreduce in hadoop
Editor to share with you how to achieve serial execution of mapreduce in hadoop, I believe most people do not know much about it, so share this article for your reference, I hope you can learn a lot after reading this article, let's go to know it!
Import java.io.IOException;import java.util.Iterator;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.NullWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer Import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class PickMain {private static final Log LOG = LogFactory.getLog (PickMain.class); public static void main (String [] args) throws IOException, ClassNotFoundException, InterruptedException {/ * * Configuration conf = new Configuration () Job job1 = Job.getInstance (conf); job1.setJarByClass (PickMain.class); job1.setMapperClass (FindMapper.class); job1.setReducerClass (FindReducer.class); job1.setOutputKeyClass (Text.class); job1.setOutputValueClass (Text.class) FileInputFormat.addInputPath (job1, new Path (args [0])); FileOutputFormat.setOutputPath (job1, new Path (args [1])); boolean flag1 = job1.waitForCompletion (true) / / the following method can also implement serial execution of job if (flag1) {Job job2 = Job.getInstance (conf); job2.setJarByClass (PickMain.class); job2.setMapperClass (SecondFindMapper.class); job2.setReducerClass (SecondFindReducer.class) Job2.setOutputKeyClass (Text.class); job2.setOutputValueClass (Text.class); FileInputFormat.addInputPath (job2, new Path (args [1])); FileOutputFormat.setOutputPath (job2, new Path (args [2])) Boolean flag2 = job2.waitForCompletion (true); System.out.println (flag2?0:1); if (flag2) {LOG.info ("The job is done!") System.exit (0);} else {LOG.info ("The Second job is wrong!"); System.exit (1) }} else {LOG.info ("The firt job is Running Wrong job break!"); System.exit (1) } * / / below: submit multiple jobs Configuration conf = new Configuration (); Job job1 = Job.getInstance (conf) by using ContolledJob and JobControl Job1.setJarByClass (PickMain.class); job1.setMapperClass (FindMapper.class); job1.setReducerClass (FindReducer.class); job1.setOutputKeyClass (Text.class); job1.setOutputValueClass (Text.class); FileInputFormat.addInputPath (job1, new Path (args [0])) FileOutputFormat.setOutputPath (job1, new Path (args [1])); Configuration conf2 = new Configuration (); Job job2 = Job.getInstance (conf2); job2.setJarByClass (PickMain.class); job2.setMapperClass (SecondFindMapper.class) Job2.setReducerClass (SecondFindReducer.class); job2.setOutputKeyClass (Text.class); job2.setOutputValueClass (Text.class); FileInputFormat.addInputPath (job2, new Path (args [1])); FileOutputFormat.setOutputPath (job2, new Path (args [2])) / / create ControlledJob to package job ControlledJob cjob1 = new ControlledJob (conf); ControlledJob cjob2 = new ControlledJob (conf2); cjob1.setJob (job1); cjob2.setJob (job2) / / sets the dependency, in which the job2 will not execute cjob2.addDependingJob (cjob1) until the job1 execution is completed; / / the JobControl class is equivalent to a job controller, which is a thread that needs to start JobControl jc = new JobControl ("my_jobcontrol") through the thread. Jc.addJob (cjob1); jc.addJob (cjob2); Thread th = new Thread (jc); th.start (); / / exit while (! jc.allFinished ()) {Thread.sleep (5000) after all job has been executed. } System.exit (0);}} class FindMapper extends Mapper {Text M1 = new Text (); Text m2 = new Text (); @ Override protected void map (LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {String line = value.toString () String [] tmp1 = line.split (":"); String outval = tmp1 [0]; String [] outkeys = tmp1 [1] .split (","); for (int I = 0; i0) {for (int I = 1bot I)
