1、MapReduce框架原理
1.1 InputFormat数据输入
1.1.1 切片与MapTask并行度决定机制
1、问题引出
MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个job的处理速度。
2、MapTask并行度决定机制
数据块：Block是HDFS物理上吧数据分成一块一块。数据块是HDFS储存数据单位。
数据切片：数据切片只是在逻辑上对输出进行分片，并不会在磁盘上将其切分成片进行储存。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。
1.1.2 Job提交流程源码与切片源码详解
1、Job提交流程源码解析
waitForCompletion() submit(); // 1 建立连接 connect(); // 1）创建提交 Job 的代理 new Cluster(getConfiguration()); // （1）判断是本地运行环境还是 yarn 集群运行环境 initialize(jobTrackAddr, conf); // 2 提交 job submitter.submitJobInternal(Job.this, cluster) // 1）创建给集群提交数据的 Stag 路径 Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf); // 2）获取 jobid ，并创建 Job 路径 JobID jobId = submitClient.getNewJobID(); // 3）拷贝 jar 包到集群 copyAndConfigureFiles(job, submitJobDir); rUploader.uploadFiles(job, jobSubmitDir); // 4）计算切片，生成切片规划文件 writeSplits(job, submitJobDir); maps = writeNewSplits(job, jobSubmitDir); input.getSplits(job); // 5）向 Stag 路径写 XML 配置文件 writeConf(conf, submitJobFile); conf.writeXml(out); // 6）提交 Job,返回提交状态 status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
Job提交流程源码解析
2、FileInputFormat切片源码解析
1.1.3 FileInputFormat 切片机制
1、切片机制
（1）简单地按照文件的内容长度进行切片
（2）切片大小。默认等于Block大小
（3）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片
2、案例分析
3、源码解析
（1）源码中计算切片大小的公式
Math.max(minSize, Math.min(maxSize, blockSize)); mapreduce.input.fileinputformat.split.minsize=1 默认值为1 mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue 因此，默认情况下，切片大小=blocksize。
（2）切片大小设置
maxsize（切片最大值）：参数如果调得比blockSize小，则会让切片变小，而且就等于配置的这个参数的值。
minsize（切片最小值）：参数调的比blockSize大，则可以让切片变得比blockSize还大。
（3）获取切片信息API
// 获取切片的文件名称 String name = inputSplit.getPath().getName(); // 根据文件类型获取切片信息 FileSplit inputSplit = (FileSplit) context.getInputSplit();
1.1.4 TextInputFormat
1、FileInputFormat实现类
FileInputFormat 常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、
NLineInputFormat、CombineTextInputFormat 和自定义 InputFormat 等。
2、TextInputFormat
TextInputFormat 是默认的 FileInputFormat 实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量， LongWritable 类型。值是这行的内容，不包括任何行终止符（换行符和回车符），Text 类型。
1.1.5 CombineTextInputFormat 切片机制
框架默认的TextInputFormat切片机制时对任务按文件规划切片，不管文件多小，都会时一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask,处理效率极其低下。
1、应用场景：
CombineTextInputFormat 用于小文件过多的场景，它可以将多个小文件从逻辑上规划到
一个切片中，这样，多个小文件就可以交给一个 MapTask 处理。
2、虚拟储存切片最大值设置
CombineTextInputFormat.setMaxInputSplitSize(job,4194304)；//4M
注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。
3、切片机制
生成切片过程包括：虚拟储存过程和切片过程两部分。
（1）虚拟储存过程：
将输入目录下所以文件大小，依次和设置的setMaxInputSpiltSize值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置·的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值两倍，此时将文件均分成2个虚拟储存块（防止出现太小切片）。
例如setMaxInputSpiltSize值为4M，输入文件大小为8.02M，先逻辑上分成一个4M。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的小的虚拟储存文件，所以将剩余的4.02文件切分成（2.01M,2.01M）两个文件。
（2）切片过程：
（a）判断虚拟储存的文件大小是否大于setMaxInputSpiltSize值，大于等于则单独形成一个切片。
（b）如果不大于则跟下一个虚拟储存文件进行合并，共同形成一个切片。
（c）测试举例： 有 4 个小文件大小分别为 1.7M、5.1M、3.4M 以及 6.8M 这四个小
文件，则虚拟存储之后形成 6 个文件块，大小分别为：1.7M，（2.55M、2.55M），3.4M 以及（3.4M、3.4M）
最终会形成三个切片，大小分别为：（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M
1.1.6 CombineTextInputFormat案例实操
1、需求
将输入的大量小文件合并成一个切片统一处理
（1）输入数据
（2）期望
期望一个切片处理四个文件
2、实现过程
（1）正常情况下，这是要生成4个切片的
（2）在 WordcountDriver 中增加如下代码，运行程序，并观察运行的切片个数为 1。
// 如果不设置 InputFormat，它默认用的是 TextInputFormat.class job.setInputFormatClass(CombineTextInputFormat.class); //虚拟存储切片最大值设置 20m CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);
运行如果为 1 个切片：
number of splits:1
说明：为什么设置20M就是只有一个切片呢？
因为四个文件加起来都没有20M大（我们设置的MaxInputSpiltSize的大小）
1.2 MapReduce 工作流程
（1）
（2）
上面的流程是整个MapReduce最全工作流程，但是Shuffle过程只是从第7步开始到第16步结束，具体Shuffle过程详解，如下：
（1）MapTask收集我们的map()方法输出的kv对，放到内存缓冲区中
（2）从内存缓冲区不断溢出到本地磁盘问就，可能会溢出多个文件
（3）多个溢出文件会被合并成大的溢出文件
（4）在溢出过程及合并过程中，都有调用Partitioner进行分区和针对key进行排序
（5）ReduceTask根据自己的分区号，去各个MapTask机器上取相对应的结果分区数据
（6）ReduceTask会抓取同一个分区的来着不同MapTask的结果文件，ReduceTask会将这些文件再进行合并（归并排序）
（7）合并成大文件后，Shuffle的过程也就结束了，后面进入ReduceTask的逻辑运算过程（从文件中取出一个一个的键值对Group，调用用户自定义的reduce()方法）
注意：
（1）Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率，从原则上·说，缓冲器越大，磁盘IO的次数越少，执行的速度越快。
（2）缓冲区的大小可以通过参数调整，参数：mapreduce.task.io.sort.mb默认是100M
