MapReduce切片与文件输入格式
MapReduce切片
MapReduce的并行度决定了Map阶段任务的并发度,但是MapTask的并行任务并不是越多越好。MapTask的并行度由切片数量决定,切片是MapReduce程序输入数据的单位,一个切片会对应一个MapTask。数据切片只是逻辑上对输入进行分片处理,不会在磁盘上对其切分成片进行处理。
FileInputFormat切片源码解析
- 找到数据存储的目录,遍历每一个文件
- 获取文件大小
- 计算切片大小:computeSplitSize()方法,计算公式为Max(minSize,min(maxSize, blockSize)) = blockSize,所以默认是block大小,要调整切片大小,可以修改maxSize和minSize,比如将maxSize调到比blockSize小,则切片大小比blockSize小,将minSize调到比blockSize大,则切片大小比blockSize大。
- 按照切片大小进行切片,每次切片时,都会片段剩下的文件是否大于切片的1.1倍,如果不大于则在一个切片中
- 将切片信息写到一个切片规划文件中,整个切片的核心过程在getSplit()方法中完成
- InputSplit只记录了切片的元数据信息,比如起始位置、长度和所在的节点等。
- 提交切片规划文件到YARN上,YARN上的MrAppMaster根据切片规划文件计算出开启的MapTask的个数。
获取切片信息的API:
TextInputFormat
FileInputFormat的常见实现接口:TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat。其中TextInputFormat是FileInputFormat的默认实现,key是LongWritable类型,代表该行的起始字节偏移量,value是Text类型,代表该行内容。其他的实现也是见名知义,比如KeyValueTextInputFormat的key和value都是Text类型,每行按照某个指定的分隔符切割为key-value键值对的形式,NLineInputFormat与TextInputFormat类似,只不过可以一次读取多行,CombineTextInputFormat则可以一次读取多个文件,用于处理多个小文件的场景,可以将多个小文件放在一个逻辑切片中,提高效率。
CombineTextInputFormat
CombineTextInputFormat的最大切片设置:CombineTextInputFormat.setMaxInputSplitSize(job, 4194304),即4M,但这并不意味着每个切片的最大大小真的就是4M,这涉及到一个虚拟存储的概念,即遇到一个大于4M的文件时,将其分为n份,使得每份小于4M,然后按照字典序排列依次相加,直到遇到大于4M则放入同一个切片中:
设置输出格式为CombineTextInputFormat:
// 如果不设置InputFormatClass,则默认为TextInputFormat.class job.setInputFormatClass(CombineTextInputFormat.class); // 设置切片最大值 CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);