1 package com.mengyao.hadoop.mapreduce;  2   3 import java.io.IOException;  4   5 import org.apache.hadoop.conf.Configuration;  6 import org.apache.hadoop.conf.Configured;  7 import org.apache.hadoop.fs.Path;  8 import org.apache.hadoop.io.NullWritable;  9 import org.apache.hadoop.io.Text; 10 import org.apache.hadoop.mapreduce.InputFormat; 11 import org.apache.hadoop.mapreduce.Job; 12 import org.apache.hadoop.mapreduce.Mapper; 13 import org.apache.hadoop.mapreduce.Reducer; 14 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 15 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat; 16 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 17 import org.apache.hadoop.util.Tool; 18 import org.apache.hadoop.util.ToolRunner; 19  20 /** 21  * KeyValueTextInputFormat默认将行的第一个制表符分隔，前面的是key，后面的是value。并且Mapper输入的k1、v1必须是 Text, Text。 22  * mapreduce.input.keyvaluelinerecordreader.key.value.separator属性可以改变分隔符，默认为“\t”。官方API描述如下： 23  *         An {
    @link InputFormat} for plain text files. Files are broken into lines. 24  *         Either line feed or carriage-return are used to signal end of line. 25  *         Each line is divided into key and value parts by a separator byte. If no 26  *         such a byte exists, the key will be the entire line and value will be empty. 27  *         The separator byte can be specified in config file under the attribute name 28  *         mapreduce.input.keyvaluelinerecordreader.key.value.separator. The default 29  *         is the tab character ('\t'). 30  *         public class KeyValueTextInputFormat extends FileInputFormat
     
       31  *          32  * 此处应用场景为获取图书大纲，找出所有的一级索引，读取输入HDFS目录下的文件/mapreduces/bookOutline.txt，内容如下： 33  *     第1章    PostgresQL服务器简介 34  *         1.1    为什么在服务器中进行程序设计 35  *         1.2    关于本书的代码示例 36  *         1.3    超越简单函数 37  *         1.4    使用触发器管理相关数据 38  *         1.5    审核更改 39  *         1.6    数据清洗 40  *         1.7    定制排序方法 41  *         1.8    程序设计最佳实践 42  *         1.8.1    KISS——尽量简单(keep it simple stupid) 43  *         1.8.2    DRY——不要写重复的代码(don't repeat yourself) 44  *         1.8.3    YAGNI——你并不需要它(you ain'tgonnaneedit) 45  *         1.8.4    SOA——服务导向架构(service-oriented architecture) 46  *         1.8.5    类型的扩展 47  *         1.9    关于缓存 48  *         1.10    总结——为什么在服务器中进行程序设计 49  *         1.10.1    性能 50  *         1.10.2    易于维护 51  *         1.10.3    保证安全的简单方法 52  *         1.11    小结 53  *     第2章    服务器程序设计环境 54  *         2.1    购置成本 55  *         2.2    开发者的可用性 56  *         2.3    许可证书 57  *         2.4    可预测性 58  *         2.5    社区 59  *         2.6    过程化语言 60  *         2.6.1    平台兼容性 61  *         2.6.2    应用程序设计 62  *         2.6.3    更多基础 63  *         2.7    小结 64  *     第3章    第一个PL／pgsQL函数 65  *         3.1    为什么是PL／pgSQL 66  *         3.2    PL／pgSQL函数的结构 67  *         ... 68  * 69  * 输出到HDFS目录下的文件/mapreduces/keyvaluetextinputformat/part-r-00000，内容如下： 70  *        第1章 71  *        第2章 72  *        第3章 73  * 74  * @author mengyao 75  * 76  */ 77 public class KeyValueTextInputFormatApp extends Configured implements Tool { 78  79     static class NLineInputFormatMapper extends Mapper
      
        { 80          81         private NullWritable outputValue; 82          83         @Override 84         protected void setup(Context context) 85                 throws IOException, InterruptedException { 86             this.outputValue = NullWritable.get(); 87         } 88          89         @Override 90         protected void map(Text key, Text value, Context context) 91                 throws IOException, InterruptedException { 92             //如果key值不为空则认为是一级索引 93             if (key.toString()!=null && !key.toString().isEmpty()) { 94                 context.write(key, this.outputValue); 95             } 96         } 97     } 98      99     static class NLineInputFormatReducer extends Reducer
       
         {100         101         private NullWritable outputValue;102         103         @Override104         protected void setup(Context context)105                 throws IOException, InterruptedException {106             this.outputValue = NullWritable.get();107         }108         109         @Override110         protected void reduce(Text key, Iterable
        
          value, Context context)111                 throws IOException, InterruptedException {112             context.write(key, outputValue);113         }114     }115     116     @Override117     public int run(String[] args) throws Exception {118         Job job = Job.getInstance(getConf(), KeyValueTextInputFormatApp.class.getSimpleName());119         job.setJarByClass(KeyValueTextInputFormatApp.class);120         121         job.setInputFormatClass(KeyValueTextInputFormat.class);122         FileInputFormat.addInputPath(job, new Path(args[0]));123         FileOutputFormat.setOutputPath(job, new Path(args[1]));124         125         job.setMapperClass(NLineInputFormatMapper.class);126         job.setMapOutputKeyClass(Text.class);127         job.setMapOutputValueClass(NullWritable.class);128         129         job.setReducerClass(NLineInputFormatReducer.class);130         job.setOutputKeyClass(Text.class);131         job.setOutputValueClass(NullWritable.class);132         133         return job.waitForCompletion(true)?0:1;134     }135 136     public static int createJob(String[] args) {137         Configuration conf = new Configuration();138         conf.set("dfs.datanode.socket.write.timeout", "7200000");139         conf.set("mapreduce.input.fileinputformat.split.minsize", "268435456");140         conf.set("mapreduce.input.fileinputformat.split.maxsize", "536870912");141         conf.set("mapreduce.job.jvm.numtasks", "-1");        142         conf.set("mapreduce.map.speculative", "false");        143         conf.set("mapreduce.reduce.speculative", "false");    144         conf.set("mapreduce.map.maxattempts", "4");            145         conf.set("mapreduce.reduce.maxattempts", "4");        146         conf.set("mapreduce.map.skip.maxrecords", "0");147         int status = 0;148         149         try {150             status = ToolRunner.run(conf, new KeyValueTextInputFormatApp(), args);151         } catch (Exception e) {152             e.printStackTrace();153         }154         155         return status;156     }157     158     public static void main(String[] args) {159         args = new String[]{"/mapreduces/bookOutline.txt", "/mapreduces/keyvaluetextinputformat"};160         if (args.length!=2) {161             System.out.println("Usage: "+KeyValueTextInputFormatApp.class.getName()+" Input paramters 
          
          
           ");162 } else {163 int status = createJob(args);164 System.exit(status);165 }166 }167 168 }