MapReduce的NLineInputFormat使用

package com.mengyao.hadoop.mapreduce;

import java.io.IOException;

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.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.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * NLineInputFormat前面的N表示每个Mapper收到输入的行数，N的默认输入行数是1。mapreduce.input.lineinputformat.linespermap属性实现N值的设定。如果希望使Mapper收到固定行数的输入可以使用该类实现。
 * 与TextInputFormat相同，key是文件中行的字节偏移量，value是行本身。
 * 通常情况下，对少量的输入行执行map任务是比较低效的（任务初始化的额外开销导致）。
 *
 * 使用NLineInputFormat设置Mapper任务每次输入处理4行，此处应用场景为获取图书大纲，找出所有的一级索引，读取输入HDFS目录下的文件/mapreduces/bookOutline.txt，内容如下：
 *     第1章    PostgresQL服务器简介
 *         1.1    为什么在服务器中进行程序设计
 *         1.2    关于本书的代码示例
 *         1.3    超越简单函数
 *         1.4    使用触发器管理相关数据
 *         1.5    审核更改
 *         1.6    数据清洗
 *         1.7    定制排序方法
 *         1.8    程序设计最佳实践
 *         1.8.1    KISS——尽量简单(keep it simple stupid)
 *         1.8.2    DRY——不要写重复的代码(don't repeat yourself)
 *         1.8.3    YAGNI——你并不需要它(you ain'tgonnaneedit)
 *         1.8.4    SOA——服务导向架构(service-oriented architecture)
 *         1.8.5    类型的扩展
 *         1.9    关于缓存
 *         1.10    总结——为什么在服务器中进行程序设计
 *         1.10.1    性能
 *         1.10.2    易于维护
 *         1.10.3    保证安全的简单方法
 *         1.11    小结
 *     第2章    服务器程序设计环境
 *         2.1    购置成本
 *         2.2    开发者的可用性
 *         2.3    许可证书
 *         2.4    可预测性
 *         2.5    社区
 *         2.6    过程化语言
 *         2.6.1    平台兼容性
 *         2.6.2    应用程序设计
 *         2.6.3    更多基础
 *         2.7    小结
 *     第3章    第一个PL／pgsQL函数
 *         3.1    为什么是PL／pgSQL
 *         3.2    PL／pgSQL函数的结构
 *         ...
 *
 * 输出到HDFS目录下的文件/mapreduces/nlineinputformat/part-r-00000，内容如下：
 *        第1章    PostgresQL服务器简介
 *        第2章    服务器程序设计环境
 *        第3章    第一个PL／pgsQL函数
 *
 * @author mengyao
 *
 */
public class NLineInputFormatApp extends Configured implements Tool {

    static class NLineInputFormatMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
        
        private Text outputValue;
        
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            this.outputValue = new Text();
        }
        
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            final String line = value.toString();
            //如果行第一个字是“第”则认为是一级索引
            if (line.startsWith("第")) {
                outputValue.set(line);
                context.write(key, this.outputValue);
            }
        }
    }
    
    static class NLineInputFormatReducer extends Reducer<LongWritable, Text, Text, NullWritable> {
        
        private Text outputKey;
        private NullWritable outputValue;
        
        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            this.outputKey = new Text();
            this.outputValue = NullWritable.get();
        }
        
        @Override
        protected void reduce(LongWritable key, Iterable<Text> value, Context context)
                throws IOException, InterruptedException {
            outputKey.set(value.iterator().next());
            context.write(this.outputKey, outputValue);
        }
    }
    
    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), NLineInputFormatApp.class.getSimpleName());
        job.setJarByClass(NLineInputFormatApp.class);
        
        NLineInputFormat.setNumLinesPerSplit(job, Integer.parseInt(args[0]));
        job.setInputFormatClass(NLineInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[1]));
        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        
        job.setMapperClass(NLineInputFormatMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);
        
        job.setReducerClass(NLineInputFormatReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.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");
        int status = 0;
        
        try {
            status = ToolRunner.run(conf, new NLineInputFormatApp(), args);
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        return status;
    }
    
    public static void main(String[] args) {
        args = new String[]{"4", "/mapreduces/bookOutline.txt", "/mapreduces/nlineinputformat"};
        if (args.length!=3) {
            System.out.println("Usage: "+NLineInputFormatApp.class.getName()+" Input paramters <LINE_NUMBER> <INPUT_PATH> <OUTPUT_PATH>");
        } else {
            int status = createJob(args);
            System.exit(status);
        }
    }

}