MapReduce的CombineFileInputFormat使用

package com.mengyao.hadoop.mapreduce;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
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.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 使用CombinerTextInputFormat可以把多个小文件打包到一个输入分片中使每个Mapper任务处理更多的数据。
 * 这样就避免了像FileInputFormat那样为每个文件（无论文件大小）都使用一个分片导致遇到小文件时会有很多的Mapper任务（每个Mapper任务处理的数据很小）。
 *
 * @author mengyao
 *
 */
public class CombinerTextInputFormatApp extends Configured implements Tool {

    static class CombinerTextInputFormatMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
        
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
        }
        
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            CombineFileSplit cfs = (CombineFileSplit)context.getInputSplit();
            String fileA = cfs.getPath(0).getName();
            String fileB = cfs.getPath(1).getName();
            Counter counterA = context.getCounter(fileA, fileA);
            Counter counterB = context.getCounter(fileB, fileB);
            if (fileA.equals(counterA.getName())) {
                counterA.increment(1L);
            }
            if (fileB.equals(counterB.getName())) {
                counterB.increment(1L);
            }
            context.write(key,value);
        }
    }
    
    static class CombinerTextInputFormatReducer extends Reducer<LongWritable, Text, LongWritable, Text> {
        
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
        }
        
        @Override
        protected void reduce(LongWritable key, Iterable<Text> value, Context context)
                throws IOException, InterruptedException {
            Iterator<Text> iterator = value.iterator();
            while (iterator.hasNext()) {
                context.write(key, iterator.next());
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), CombinerTextInputFormatApp.class.getSimpleName());
        job.setJarByClass(CombinerTextInputFormatApp.class);
        
        job.setInputFormatClass(CombineTextInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        job.setMapperClass(CombinerTextInputFormatMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);
        
        job.setReducerClass(CombinerTextInputFormatReducer.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);
        
        return job.waitForCompletion(true)?0:1;
    }
    
    public static int createJob(String[] args) {
        Configuration conf = new Configuration();
        conf.set("dfs.datanode.socket.write.timeout", "7200000");
        conf.set("mapreduce.input.fileinputformat.split.minsize", "268435456");
        conf.set("mapreduce.input.fileinputformat.split.maxsize", "536870912");
        conf.set("mapreduce.job.jvm.numtasks", "-1");        
        conf.set("mapreduce.map.speculative", "false");        
        conf.set("mapreduce.reduce.speculative", "false");    
        conf.set("mapreduce.map.maxattempts", "4");            
        conf.set("mapreduce.reduce.maxattempts", "4");        
        conf.set("mapreduce.map.skip.maxrecords", "0");
        int status = 0;
        
        try {
            status = ToolRunner.run(conf, new CombinerTextInputFormatApp(), args);
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        return status;
    }
    
    public static void main(String[] args) {
        args = new String[]{"/mapreduces/*.txt", "/mapreduces/combinertextinputFormat"};
        if (args.length!=2) {
            System.out.println("Usage: "+CombinerTextInputFormatApp.class.getName()+" Input paramters <INPUT_PATH> <OUTPUT_PATH>");
        } else {
            int status = createJob(args);
            System.exit(status);
        }
    }
    
}