maptask的并行度

maptask的并行度

maptask&＃xff1a;执行map阶段的任务称为maptask。
并行度&＃xff1a;maptask一起执行的个数 换句话说在一个job中就是maptask运行的个数。

maptask运行的并行度&＃xff1a;

底层的数据如何存储的&＃xff1a;分块多副本存储的
300M文件
blk1:0-127M
blk2:128-255M
blk3:256-300M
maptask运行的个数 和数据的大小有关
一个maptask----> 对应的数据量&＃xff1f;
130M—> 数据的跨节点传输的问题
100M—> 数据存在跨节点传输的问题
1个maptask-----> 一个数据块的大小 128M
事实上一个maptask对应的数据是一个逻辑切片的大小
一个逻辑切片---- 128M 一个块的大小
逻辑切片&＃xff1a;逻辑上的概念 一个数据偏移范围的划分 并没有真正的切分数据
物理切片&＃xff1a;是逻辑切片的对立面
一个逻辑切片对应一个maptask任务

底层源码&＃xff1a;

FileInputFormat 文件加载器
getSplits方法 获取所有的切片

/** * Generate the list of files and make them into FileSplits.* &＃64;param job the job context* &＃64;throws IOException* 切片获取的方法* InputSplit 代表文件切片对象* 返回值&＃xff1a;List 所有文件切片的列表* 参数&＃xff1a;JobContext job job的上下文对象*/public List getSplits(JobContext job) throws IOException {StopWatch sw &＃61; new StopWatch().start();//获取最小切片大小/*getFormatMinSplitSize() &＃xff1a;1getMinSplitSize(job)&＃xff1a;0minSize &＃61;1*/long minSize &＃61; Math.max(getFormatMinSplitSize(), getMinSplitSize(job));//切片最大大小long maxSize &＃61; getMaxSplitSize(job);// generate splits//最终返回的list集合List splits &＃61; new ArrayList();//获取FileInputFormat.addInputPath("")输入路径下的所有文件状态信息对象&＃xff0c;包含块大小信息List files &＃61; listStatus(job);//循环遍历每一个文件的状态信息for (FileStatus file: files) {Path path &＃61; file.getPath();//获取文件的实际长度long length &＃61; file.getLen();if (length !&＃61; 0) {BlockLocation[] blkLocations;if (file instanceof LocatedFileStatus) {blkLocations &＃61; ((LocatedFileStatus) file).getBlockLocations();} else {FileSystem fs &＃61; path.getFileSystem(job.getConfiguration());blkLocations &＃61; fs.getFileBlockLocations(file, 0, length);}//判断是否可以逻辑切片if (isSplitable(job, path)) {//获取配置的块大小&＃xff0c;默认128Mlong blockSize &＃61; file.getBlockSize();//核心 计算每一个切片大小的/*blockSize&＃xff1a;128MminSize&＃xff1a;1maxSize&＃xff1a;Long.MAX_VALUE*/long splitSize &＃61; computeSplitSize(blockSize, minSize, maxSize);//计算切片剩余大小long bytesRemaining &＃61; length;//切片/*是否还进行切分的条件SPLIT_SLOP //切片的临界值 &＃61;1.1剩余大小/128M > 1.1剩余大小>128M * 1.1 140&＃43;((double) bytesRemaining)/splitSize > SPLIT_SLOP*/while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {int blkIndex &＃61; getBlockIndex(blkLocations, length-bytesRemaining);splits.add(makeSplit(path, length-bytesRemaining, splitSize,blkLocations[blkIndex].getHosts(),blkLocations[blkIndex].getCachedHosts()));bytesRemaining -&＃61; splitSize;}if (bytesRemaining !&＃61; 0) {int blkIndex &＃61; getBlockIndex(blkLocations, length-bytesRemaining);splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,blkLocations[blkIndex].getHosts(),blkLocations[blkIndex].getCachedHosts()));}} else { // not splitablesplits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),blkLocations[0].getCachedHosts()));}} else { //Create empty hosts array for zero length filessplits.add(makeSplit(path, 0, length, new String[0]));}}// Save the number of input files for metrics/loadgenjob.getConfiguration().setLong(NUM_INPUT_FILES, files.size());sw.stop();if (LOG.isDebugEnabled()) {LOG.debug("Total # of splits generated by getSplits: " &＃43; splits.size()&＃43; ", TimeTaken: " &＃43; sw.now(TimeUnit.MILLISECONDS));}return splits;}


//最小切片大小
/*** Get the minimum split size* &＃64;param job the job* &＃64;return the minimum number of bytes that can be in a split* job.getConfiguration()&＃xff1a;获取job的所有配置文件信息,写到 job.xml中 * job.settJarByClass(Driver.class);jar.classDriver.class&＃xff0c;没有配置的则使用默认值* getLong(配置属性名&＃xff0c;默认返回值)&＃xff0c;先去配置文件中获取配置属性的值&＃xff0c;获取到则返回&＃xff0c;获取不到则返回默认值&＃xff08;参数2&＃xff09;。* mapreduce.input.fileinputformat.split.minsize,默认是&＃xff1a;0* mapred-default.xml ------> return 0*r*/public static long getMinSplitSize(JobContext job) {return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);}


//最大切片大小
/*** Get the maximum split size.* &＃64;param context the job to look at.* &＃64;return the maximum number of bytes a split can include* */public static long getMaxSplitSize(JobContext context) {/*** context.getConfiguration()&＃xff0c;获取job.xml* mapreduce.input.fileinputformat.split.maxsize,默认没有设置* return Long.MAX_VALUE*/return context.getConfiguration().getLong(SPLIT_MAXSIZE, Long.MAX_VALUE);}


//计算切片大小/*Math.max(minSize,blockSize)return blockSize 默认一个切片的大小就是一个块的大小调整切片大小大于128M ----->调整minSize小于128M ----->调整maxSize具体操作&＃xff1a;//不推荐使用1&＃xff09;修改配置文件&＃xff1a;mapred-site.xml大于128Mmapreduce.input.fileinputformat.split.minsize130*1024*1024The minimum size chunk that map input should be splitinto. Note that some file formats may have minimum split sizes thattake priority over this setting.小于128Mmapreduce.input.fileinputformat.split.maxsize100*1024*1024The minimum size chunk that map input should be splitinto. Note that some file formats may have minimum split sizes thattake priority over this setting.2&＃xff09;代码中修改//修改大于128MFileInputFormat.setMinInputSplitSize(job, size);//修改小于128MFileInputFormat.setMaxInputSplitSize(job, size);注意:1)一个文件如果不大于128M&＃xff0c;这时候单独成一个切片文件大于128M&＃xff0c;按照分块来2)文件切片划分的时候&＃xff0c;最后一个切片有可能大于128M&＃xff0c;128*1.1268M存储blk01 1-128Mblk02 129-256M 128Mblk03 257-268M 12M启动maptask时候 切片划分&＃xff1a;split01 268/128>1.1 1-128剩余 268-128&＃61;140Msplit02 140/128<1.1 不切 129-268M跨块或跨节点 比单独启动一个maptask划算*/protected long computeSplitSize(long blockSize, long minSize,long maxSize) {return Math.max(minSize, Math.min(maxSize, blockSize));}


合并小文件

默认的FileInputFormat的实现&＃xff1a;TextInputFormat,这种方式 一个文件读一次&＃xff0c;不会进行合并。
///逻辑合并&＃xff0c;maptask的输入&＃xff0c;将多个小文件合并到一个maptask中&＃xff0c;根据切片大小进行合并的 &＃xff0c;逻辑切片进行合并的时候&＃xff0c;并不会进行真正的合并&＃xff0c;
//合并的时候&＃xff0c;根据切片的大小进行合并文件&＃xff0c;输入的时候多喝小文件启动一个maptask任务。
job.setInputFormatClass(CombineTextInputFormat.class);
//设置切片大小 >128M
CombineTextInputFormat.setMinInputSplitSize(job, 13010241024);
//指定输入&＃xff0c;设置切片大小
FileInputFormat.setMinInputSplitSize(job, 13010241024);

总结&＃xff1a;一个maptask------一个yarnchild--------一个逻辑切片的数据------>默认大小block的大小。

逻辑切块和切块的区别&＃xff1a;
1&＃xff09;逻辑切片是逻辑概念&＃xff0c;范围的划分
切块是真实的物理切分
2&＃xff09;逻辑切片针对的是mapreduce运行过程&＃xff0c; 对应的是maptask
切块针对于数据的存储