map函数处理的<k1,v1>从HDFS文件中获取流程源码解析
0 引子：
mapreduce在执行任务的时候，是如何将外部文件进行切分，并将这些文件转换成<k1,v1>键值对方式的
（还记得 map-reduce基本概念和wordcount解析 文章中提到的<k1,v1>概念吗?）
一些总结性的话：
a) recordreader + inputsplit是数据输入处理阶段非常重要的两个概念。
b) inputsplit: 对原始输入数据的封装，封装原始数据源，这个数据源可以是hdfs文件系统，也可是DB,或者数据流
c) recordreader： 把inputsplit的数据转换成<k1,v1>
d) inputsplit和hdfs的block之间的关系是什么？？
在hdfs看来，mapreduce就是hdfs的客户端，
mapreduce默认inputsplit大小和hdfs的一个block相等，
但是并不说
mapreduce的一个inputsplit就对应hdfs的一个block,
mapreduce是看不到hdfs体内的一个个的block的, 因此默认情况下，被处理目标文件有多少block，就会产生
多少inputsplit,也就会对应多少个map任务，即block和inputsplit只有这种对照关系。
为什么设计成一个hdfs block 对应一个 inputsplit:
核心就是数据本地化,在当前机器上计算的数据尽量不要传输
1个inputsplit>1个block时，比如1个inputsplit使用2个block时,hdfs的这两个block不一定都存储在同一个节点上,那么必然会产生网络传输，将需要的另一个block 传输到计算的这个节点上.
inputsplit<1个block时,那么一个block的数据中就必然有一部分没有被处理，必然就会被别的mapper处理，必然增大网络传输的概率.
看下图：
1 FileInputFormat代码简析, 回答如何将源文件逻辑分割
package org.apache.hadoop.mapreduce.lib.input
读取hdfs目录下的文件，后将文件便利，在执行getSplits方法时会发现，for循环内代码中，
每一个文件不论是1K还是1G
都会被处理封装成对应个数的inputsplit ，即一个文件对应一个或者多个split
而每一个split都会对应一个map进程
如果小文件太多，那么开启map进程太多 势必浪费资源，操作系统开启，执行，挂起一个进程是很消耗资源的
/**
* Generate the list of files and make them into FileSplits.
*/
public List<InputSplit> getSplits(JobContext job
) throws IOException {
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
// generate splits
List<InputSplit> splits = new ArrayList<InputSplit>();
// hdfs//master:9000/hello 则files为hello hdfs//master:9000/ 则files为目录下的所有文件
List<FileStatus>files = listStatus(job); // 读取job中的输入路径，进入
for (FileStatus file: files) {// 对目标路径下的文件/文件数 进行截取成 split操作， 每个文件至少对应一个切片，一个切片对应一个map
//如果每个文件都是1M,那么这个文件就在切片时对应一个split 也就对应一个map操作，1T的文件就1T个map，这种计算要消费多大的资源，这就是小文件不适合hdfs的原因，
//而如果64个1m文件合并后在处理，对应map就缩小了63倍，这样对程序性能影响有多少 可想而知
Path path = file.getPath();// 获取文件路径
FileSystem fs = path.getFileSystem(job.getConfiguration());// 获取hdfs操作
long length = file.getLen();
BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
if ((length != 0) && isSplitable(job, path)) { // 当目标文件长度>0并且是可split时
long blockSize = file.getBlockSize();// 文件块大小 默认是64M
long splitSize = computeSplitSize(blockSize, minSize, maxSize);// 计算切片尺寸，进入此方法，可见结果为64M, 参数minSize值为1，maxSize为long的最大值
如果想调整切片大小，需要重写方法 getMaxSplitSize public static long getMinSplitSize( 两个方法
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(new FileSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkLocations.length-1].getHosts()));
}
} else if (length != 0) { // 文件不可拆分
splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
} else { // 文件长度=0
//Create empty hosts array for zero length files
splits.add(new FileSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files in the job-conf
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
LOG.debug("Total # of splits: " + splits.size());
return splits;
}
如下：
protected List<FileStatus> listStatus(
....
Path[] dirs = getInputPaths(job); // 进入 大概看下 ）
仅仅是对1T数据按照64M一个个的做标记， 是逻辑切分，不是物理切分，仅仅记录原始数据位置和要处理的长度
类比于高速公路上的 37公里处， 70公里处的路标， 仅仅是个标志而已，实际上高速公路还是连在一起的
看下面的数据结构，如果是物理切分的话，其数据结构必然也会要有 private bytes[] buf; 这个属性
public class FileSplit extends InputSplit implements Writable {
private Path file;
private long start;
private long length;
private String[] hosts;
。。。。
}2 RecordReader 代码简析，回答如何将split转换成键值对:
package org.apache.hadoop.mapreduce.lib.input
TextInputFormat.
@Override
public RecordReader<LongWritable, Text>
createRecordReader(InputSplit split,
TaskAttemptContext context) {
return new LineRecordReader(); // 进入
}
}
如下：
/**
* Treats keys as offset in file and value as line.
*/
LineRecordReader.java
nextKeyValue() 调用一次 则当前 key value 就被赋值到， 相当于迭代器
public boolean nextKeyValue() throws IOException {
if (key == null) {
key = new LongWritable(); // 第一次调用时，key = 0 eg: hello.txt (大小10KB)
}
key.set(pos);
if (value == null) {
value = new Text();
}
int newSize = 0;
// We always read one extra line, which lies outside the upper
// split limit i.e. (end - 1)
while (getFilePosition() <= end) {
newSize = in.readLine(value, maxLineLength, //读取第一行时，将第一行内容写入到value内，并返回第一行读完后的位置
Math.max(maxBytesToConsume(pos), maxLineLength));
if (newSize == 0) {
break;
}
pos += newSize; // 记录pos现在的位置，为读取第二行时， key.set(pos);做准备工作
if (newSize < maxLineLength) {
break;
}
// 这样赋值对全局变量 key value赋值完毕后，通过getCurrentKey getCurrentValue 获取现在的值
....}
看map如何调用写法：
Mapper.java:
public void run(Context context) throws IOException, InterruptedException {
setup(context);
while (context.nextKeyValue()) {// 不断context.nextKeyValue()进入方法后实际就是RecordReader.nextKeyValue()来不断将目标文件的行，行内容放在<k,v>
map(context.getCurrentKey(), context.getCurrentValue(), context); //不断取值
}
cleanup(context);
}总结如下：
系统回答如下：
问：从源代码的角度分析map函数处理的<k1,v1>是如何从HDFS文件中获取的？ 答： 1.从TextInputFormat入手分析，找到父类FileInputFormat，找到父类InputFormat。 在InputFormat中找到2个方法，分别是getSplits(...)和createRecordReader(...)。 通过注释知道getSplits(...)作用是把输入文件集合中的所有内容解析成一个个的InputSplits，每一个InputSplit对应一个mapper task。 createRecordReader(...)作用是创建一个RecordReader的实现类。RecordReader作用是解析InputSplit产生一个个的<k,v>。 2.在FileInputFormat中找到getSplits(...)的实现。 通过实现，获知 (1)每个SplitSize的大小和默认的block大小一致，好处是满足数据本地性。 (2)每个输入文件都会产生一个InputSplit，即使是空白文件，也会产生InputSPlit； 如果一个文件非常大，那么会按照InputSplit大小，切分产生多个InputSplit。 3.在TextInputFormat中找到createRecordReader(...)的实现，在方法中找到了LineRecordReader。 接下来分析LineRecordReader类。 在RecordReader类中，通过查看多个方法，知晓key、value作为类的属性存在的，且知道了nextKeyValue()方法的用法。 在LineRecordReader类中，重点分析了nextKeyValue(...)方法。在这个方法中，重点分析了newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos)); 在in.readLine(...)中，第一个形参存储被读取的行文本内容，返回值表示被读取内容的字节数。 通过以上代码，分析了InputSplit中的内容是如何转化为一个个的<k,v>。 4.从Mapper类中进行分析，发现了setup()、cleanup()、map()、run()。 在run()方法中，通过while，调用context.nextKeyValue(...)。 进一步分析Context的接口类是org.apache.hadoop.mapreduce.lib.map.WrappedMapper.MapContext，MapContext调用了nextKeyValue(...)。最终找到了MapContext的实现了MapContextImpl类org.apache.hadoop.mapreduce.task.MapContextImpl。 在这个类的构造方法中，发现传入了RecordReader的实现类。