[TOC] # 简介 MapReduce任务的输入文件一般是存储在HDFS里面.输入的文件格式包括:基于行的日志文件,二进制格式文件.这些文件一般很大,达到数十G,甚至更大,那么MapReduce是如何读取这些数据的? InputFormat常见的接口实现类包括: TextInputFormat, KeyValueTextInputFormat, NLineInputFormat, CombineTextInputFormat和自定义InputFormat等 # TextInputFormat TextInputFormat是默认的InputFormat.每条记录是一行输入. 键是LongWritable类型,存储该行在整个文件中的字节偏移量,值是这行的内容,不包括任何行终止符(换行符和回车符) 一些是一个示例,比如一个分片包含了如下4条文本记录 ~~~ Rich learning form Intelligent learning engine Learning more convenient From the real demand for more close to the enterprise ~~~ 每条记录表示为以下键值对 ~~~ (0, Rich learning form) (19, Intelligent learning engine) (47, Learning more convenient) (72, From the real demand for more close to the enterprise) ~~~ 很明显,键并不是行号.一般情况下,很难取得行号,因为文件按字节而不是按行切分为分片 # KeyValueTextInputFormat **每一行均为一条记录**,被分割符分割成key, value. 可以通过在驱动类中设置 ~~~ conf.set(keyValueLineRecordReader.KEY_VALUE_SEPERATOR, " "); ~~~ 来设定分割符,默认分割符是tab(\t) 以下是一个示例,输入是一个包含4条记录的分片.其中--->表示一个(水平方向的)制表符 ~~~ line1 ---> Rich learning form line2 ---> Intelligent learning engine line3 ---> Learning more convenient line4 ---> From the real demand for more close to the enterprise ~~~ 每条记录表示为以下键/值对 ~~~ (line1, Rich learning form) (line2, Intelligent learning engine) (line3, Learning more convenient) (line4, From the real demand for more close to the enterprise) ~~~ 此时的键是每行排在制表符之前的Text序列 ## 代码 **map** ~~~ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class KVTextMapper extends Mapper<Text, Text, Text, LongWritable> { Text k = new Text(); LongWritable v = new LongWritable(); @Override protected void map(Text key, Text value, Context context) throws IOException, InterruptedException { //设置key和value k.set(key); //设置key的个数 v.set(1); //写出 context.write(k, v); } } ~~~ **reducer** ~~~ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class KVTextReducer extends Reducer<Text, LongWritable, Text, LongWritable> { LongWritable v = new LongWritable(); @Override protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException { long count = 0l; //汇总统计 for (LongWritable value : values) { count += value.get(); } v.set(count); //输出 context.write(key, v); } } ~~~ **驱动** ~~~ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader; import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class MyDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); //设置切割符 conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " "); //获取job对象 Job job = Job.getInstance(conf); //设置jar包关系,关联mapper和reducer job.setJarByClass(MyDriver.class); //告诉框架,我们程序所用的mapper类和reduce类是什么 job.setMapperClass(KVTextMapper.class); job.setReducerClass(KVTextReducer.class); //告诉框架我们程序输出的类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); //告诉框架,我们程序使用的数据读取组件,结果输出所用的组件是什么 //TextInputFormat是mapreduce程序中内置的一种读取数据组件,准备的叫做读取文本的输入组件 job.setInputFormatClass(KeyValueTextInputFormat.class); //job.setOutputFormatClass(TextOutputFormat.class); //告诉框架,我们要处理的数据文件在那个路径下 FileInputFormat.setInputPaths(job,new Path("/Users/jdxia/Desktop/website/data/input/")); //告诉框架我们的处理结果要输出到什么地方 FileOutputFormat.setOutputPath(job,new Path("/Users/jdxia/Desktop/website/data/output/")); //提交后,然后等待服务器端返回值,看是不是true boolean res = job.waitForCompletion(true); //设置成功就退出码为0 System.exit(res?0:1); } } ~~~ # NLineInputFormat 如果使用NLineInputFormat,代表每个map进程处理的InputSplit不再按block块去划分,而是按NLineInputFormat指定的行数N来划分.即输入文件的总行数/N=切片数,如果不整除,切片数=商+1 以下是一个示例,仍然以上面的4行输入为例子 ~~~ Rich learning form Intelligent learning engine Learning more convenient From the real demand for more close to the enterprise ~~~ 例如,如果N是2,则每个输入分片包含两行,开启2个maptask ~~~ (0, Rich learning form) (19, Intelligent learning engine) ~~~ 另一个mapper则收到后两行 ~~~ (47, Learning more convenient) (72, From the real demand for more close to the enterprise) ~~~ 这里的键和值与TextInputFormat生成的一样 ## 代码 **map** ~~~ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class NLineMapper extends Mapper<LongWritable, Text, Text, LongWritable> { private Text k = new Text(); private LongWritable v = new LongWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //获取一行 String line = value.toString(); //切割 String[] splited = line.split(" "); //循环写出 for (int i = 0; i < splited.length; i++) { k.set(splited[i]); context.write(k, v); } } } ~~~ **reducer** ~~~ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class NLineReducer extends Reducer<Text, LongWritable, Text, LongWritable> { LongWritable v = new LongWritable(); @Override protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException { long count= 0l; //汇总 for (LongWritable value : values) { count += value.get(); } v.set(count); //输出 context.write(key, v); } } ~~~ **驱动** ~~~ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; 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 java.io.IOException; public class NLineDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //获取job对象 Configuration conf = new Configuration(); Job job = Job.getInstance(conf); //设置每个切片InputSplit中划分三条记录 NLineInputFormat.setNumLinesPerSplit(job, 3); //使用NLineInputFormat处理记录 job.setInputFormatClass(NLineInputFormat.class); //设置jar包位置,关联mapper和reducer job.setJarByClass(NLineDriver.class); job.setMapperClass(NLineMapper.class); job.setReducerClass(NLineReducer.class); //设置map输出kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); //设置最终输出kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); //告诉框架,我们要处理的数据文件在那个路径下 FileInputFormat.setInputPaths(job,new Path("/Users/jdxia/Desktop/website/data/input/")); //告诉框架我们的处理结果要输出到什么地方 FileOutputFormat.setOutputPath(job,new Path("/Users/jdxia/Desktop/website/data/output/")); //这边不用submit,因为一提交就和我这个没关系了,我这就断开了就看不见了 // job.submit(); //提交后,然后等待服务器端返回值,看是不是true boolean res = job.waitForCompletion(true); //设置成功就退出码为0 System.exit(res?0:1); } } ~~~ 输出结果的切片数 ~~~ number of splits ~~~