学习记录CombineFileInputFormat类

Posted 2020-12-22 zyanrong

基本摘自网上给自己做一记录，如有错误请斧正

CombineFileInputFormat是一个抽象类。Hadoop提供了两个实现类CombineTextInputFormat和CombineSequenceFileInputFormat。

其继承关系如下：

InputFormat
--->FileInputFormat
------>CombineFileInputFormat

CombineTextInputFormat的使用：

仅输入一个文件路径时：

//指定输入格式CombineFileInputFormat
job.setInputFormatClass(CombineTextInputFormat.class); 
//指定SplitSize
CombineTextInputFormat.setMaxInputSplitSize(job, 60*1024*1024L);
//指定输入路径
CombineTextInputFormat.addInputPath(job, new Path(args[0]));

对于多路径输入：多次调用，加载不同路径

//指定输入格式CombineFileInputFormat
job.setInputFormatClass(CombineTextInputFormat.class); 
//指定SplitSize
CombineTextInputFormat.setMaxInputSplitSize(job, 128*1024*1024L);
//指定输入路径
CombineTextInputFormat.addInputPath(job, new Path(args[0]));
CombineTextInputFormat.addInputPath(job, new Path(args[1]));

一次调用，同时加载多路径(字符串用逗号隔开)：

//这种方式的第二个参数只能是:将路径以逗号拼接的字符串
FileInputFormat.addInputPaths(job,"hdfs://mine:9000/orc/can,hdfs://mine:9000/orc/gps");

多路径不同类型输入：

//指定SplitSize
CombineTextInputFormat.setMaxInputSplitSize(job, 128*1024*1024L);
//指定输入路径,以及指定输入格式
MultipleInputs.addInputPath(job, new Path(args[0]), CombineTextInputFormat.class);
MultipleInputs.addInputPath(job, new Path(args[1]), CombineSequenceFileInputFormat.class);

以上两种多路径输入方式处理切割：

CombineTextInputFormat 是先把所有的输入集中起来求出总的输入大小，再除以SplitSize算出总的map个数

MultipleInputs 是先分别算出每个MultipleInputs路径对应的map个数，再对两个MultipleInputs的map个数求和

CombineFileInputFormat的使用：

常用属性值：

1.文件指定切片大小的最大值， CombineFileInputFormat.setMaxInputSplitSize()方法进行设置【不设置,则所有输入只启动一个map任务】；

2.同一节点的数据块形成切片时，切片大小的最小值，CombineFileInputFormat.setMinSplitSizeNode()方法进行设置；

3.同一机架的数据块形成切片时，切片大小的最小值，CombineFileInputFormat.setMinSplitSizeRack()方法进行设置

注意：　　

　　1）、文件切割最小值默认是1，最大值默认是long的最大值，这个不用设置最小值即可；

　　2）、存在关系：文件切割数> 同一个节点文件个数最小值 > 同一个机架上的文件个数最小值

切片的形成过程：

第一个过程： 不断迭代节点列表，逐个节点形成切片(Local Split)

　　a. 如果maxSplitSize == 0，则整个节点上的Block数据形成一个切片

　　b. 如果maxSplitSize != 0，遍历并累加每个节点上的数据块，如果累加数据块大小 >= maxSplitSize，则将这些数据块形成一个切片。继续该过程，直到剩余数据块累加大小 < maxSplitSize 。则进行下一步

　　c. 如果剩余数据块累加大小 >= minSplitSizeNode，则将这些剩余数据块形成一个切片。继续该过程，直到剩余数据块累加大小 < minSplitSizeNode。然后进行下一步，并这些数据块留待后续处理

 

第二个过程： 不断迭代机架列表，逐个机架形成切片(Rack Split)

　　a. 遍历并累加这个机架上所有节点的数据块 (这些数据块即上一步遗留下来的数据块)，如果累加数据块大小 >= maxSplitSize，则将这些数据块形成一个切片。继续该过程，直到剩余数据块累加大小<maxSplitSize。则进行下一步

　　b. 如果剩余数据块累加大小 >= minSplitSizeRack，则将这些剩余数据块形成一个切片。如果剩余数据块累加大小 < minSplitSizeRack，则这些数据块留待后续处理     

第三个过程：遍历并累加所有Rack上的剩余数据块，如果累加数据块大小 >= maxSplitSize，则将这些数据块形成一个切片。继续该过程，直到剩余数据块累加大小< maxSplitSize。则进行下一步

 

第四个过程：将最终剩余的数据块形成一个切片。

 

举个带数据的栗子说明下：

规定：maxSplit=100 > minSizeNode=50 > minSizeRack=30

原有文件：　　　　　　　　　　　　　Rack01：{[30,60,70] [80,110]} 　　Rack02：{170}　　

 

处理过程：

节点数据块判断：　　　　　　　　　　30+60+70 > 100 ? 100+60　　80+110 > 100 ? 100+90　　170 > 100 ? 100+70　　

　　　　　　　　　　　　　　　　　　--->　　3个数据切片，以及Rack01：{[60] [90]}　　Rack02：{70}　　

节点所有剩余数据累加判断节点切割：　　　--->　　60 > 50 ? 50+10　　90 > 50 ? 50+40　　70 > 50 ? 50+20　　

　　　　　　　　　　　　　　　　　　　　　　--->　　3+3个数据切片，以及Rack01：{[10] [40]}　　Rack02：{20}　　

机架迭代所有剩余累加判断：　　　　　　　　　　　--->　　10+40 < 100 ?0　　20 < 100 ? 0　　

　　　　　　　　　　　　　　　　　　　　　　　　　　--->　　3+3+0个数据切片，以及Rack01：{50}　　Rack02：{20}　　

机架所有剩余数据累加判断机架切割：　　　　　　　　　　　--->　　50+20 > 30 ? 30+30+10　　

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　--->　　3+3+0+3个数据切片

总结记录：1.节点先内部进行文件大小判断切割，再节点判断切割，2.机架内部进行文件大小判断切割，再机架判断切割