Hadoop: MapReduce2多個job串行處理

復雜的MapReduce處理中，往往需要將復雜的處理過程，分解成多個簡單的Job來執行，第1個Job的輸出做為第2個Job的輸入，相互之間有一定依賴關系。以上一篇中的求平均數為例，可以分解成三個步驟：

1. 求Sum

2. 求Count

3. 計算平均數

每1個步驟看成一個Job，其中Job3必須等待Job1、Job2完成，並將Job1、Job2的輸出結果做為輸入，下面的代碼演示了如何將這3個Job串起來

package yjmyzz.mr.job.link;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
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.output.FileOutputFormat;
import yjmyzz.util.HDFSUtil;

import java.io.IOException;


public class Avg2 {

    private static final Text TEXT_SUM = new Text("SUM");
    private static final Text TEXT_COUNT = new Text("COUNT");
    private static final Text TEXT_AVG = new Text("AVG");

    //計算Sum
    public static class SumMapper
            extends Mapper<LongWritable, Text, Text, LongWritable> {

        public long sum = 0;

        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            sum += Long.parseLong(value.toString());
        }

        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(TEXT_SUM, new LongWritable(sum));
        }

    }

    public static class SumReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

        public long sum = 0;

        public void reduce(Text key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            for (LongWritable v : values) {
                sum += v.get();
            }
            context.write(TEXT_SUM, new LongWritable(sum));
        }

    }

    //計算Count
    public static class CountMapper
            extends Mapper<LongWritable, Text, Text, LongWritable> {

        public long count = 0;

        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            count += 1;
        }

        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(TEXT_COUNT, new LongWritable(count));
        }

    }

    public static class CountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

        public long count = 0;

        public void reduce(Text key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            for (LongWritable v : values) {
                count += v.get();
            }
            context.write(TEXT_COUNT, new LongWritable(count));
        }

    }

    //計算Avg
    public static class AvgMapper
            extends Mapper<LongWritable, Text, LongWritable, LongWritable> {

        public long count = 0;
        public long sum = 0;

        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] v = value.toString().split("\t");
            if (v[0].equals("COUNT")) {
                count = Long.parseLong(v[1]);
            } else if (v[0].equals("SUM")) {
                sum = Long.parseLong(v[1]);
            }
        }

        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(new LongWritable(sum), new LongWritable(count));
        }

    }


    public static class AvgReducer extends Reducer<LongWritable, LongWritable, Text, DoubleWritable> {

        public long sum = 0;
        public long count = 0;

        public void reduce(LongWritable key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            sum += key.get();
            for (LongWritable v : values) {
                count += v.get();
            }
        }

        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(TEXT_AVG, new DoubleWritable(new Double(sum) / count));
        }

    }


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

        Configuration conf = new Configuration();

        String inputPath = "/input/duplicate.txt";
        String maxOutputPath = "/output/max/";
        String countOutputPath = "/output/count/";
        String avgOutputPath = "/output/avg/";

        //刪除輸出目錄(可選,省得多次運行時,總是報OUTPUT目錄已存在)
        HDFSUtil.deleteFile(conf, maxOutputPath);
        HDFSUtil.deleteFile(conf, countOutputPath);
        HDFSUtil.deleteFile(conf, avgOutputPath);

        Job job1 = Job.getInstance(conf, "Sum");
        job1.setJarByClass(Avg2.class);
        job1.setMapperClass(SumMapper.class);
        job1.setCombinerClass(SumReducer.class);
        job1.setReducerClass(SumReducer.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(LongWritable.class);
        FileInputFormat.addInputPath(job1, new Path(inputPath));
        FileOutputFormat.setOutputPath(job1, new Path(maxOutputPath));


        Job job2 = Job.getInstance(conf, "Count");
        job2.setJarByClass(Avg2.class);
        job2.setMapperClass(CountMapper.class);
        job2.setCombinerClass(CountReducer.class);
        job2.setReducerClass(CountReducer.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(LongWritable.class);
        FileInputFormat.addInputPath(job2, new Path(inputPath));
        FileOutputFormat.setOutputPath(job2, new Path(countOutputPath));


        Job job3 = Job.getInstance(conf, "Average");
        job3.setJarByClass(Avg2.class);
        job3.setMapperClass(AvgMapper.class);
        job3.setReducerClass(AvgReducer.class);
        job3.setMapOutputKeyClass(LongWritable.class);
        job3.setMapOutputValueClass(LongWritable.class);
        job3.setOutputKeyClass(Text.class);
        job3.setOutputValueClass(DoubleWritable.class);

        //將job1及job2的輸出為做job3的輸入
        FileInputFormat.addInputPath(job3, new Path(maxOutputPath));
        FileInputFormat.addInputPath(job3, new Path(countOutputPath));
        FileOutputFormat.setOutputPath(job3, new Path(avgOutputPath));

        //提交job1及job2,並等待完成
        if (job1.waitForCompletion(true) && job2.waitForCompletion(true)) {
            System.exit(job3.waitForCompletion(true) ? 0 : 1);
        }

    }


}

輸入文本在上一篇可以找到，上面這段代碼的主要思路：

1. Sum和Count均采用相同的輸入/input/duplicate.txt，然后將各自的處理結果分別輸出到/output/max/及/output/count/下

2. Avg從/output/max及/output/count獲取結果做為輸入，然后根據Key值不同，拿到sum和count的值，最終計算並輸出到/output/avg/下