本文共 14843 字,大约阅读时间需要 49 分钟。
在MRAppMaster中,当MapReduce作业初始化时,它会通过作业状态机JobImpl中InitTransition的transition()方法,进行MapReduce作业初始化相关操作,而这其中就包括:
1、调用createSplits()方法,创建分片,并获取任务分片元数据信息TaskSplitMetaInfo数组taskSplitMetaInfo;
2、确定Map Task数目numMapTasks:分片元数据信息数组的长度,即有多少分片就有多少numMapTasks;
3、确定Reduce Task数目numReduceTasks,取作业参数mapreduce.job.reduces,参数未配置默认为0;
4、根据分片元数据信息计算输入长度inputLength,也就是作业大小;
5、根据作业大小inputLength,调用作业的makeUberDecision()方法,决定作业运行模式是Uber模式还是Non-Uber模式。
相关关键代码如下:
// 调用createSplits()方法,创建分片,并获取任务分片元数据信息TaskSplitMetaInfo数组taskSplitMetaInfo TaskSplitMetaInfo[] taskSplitMetaInfo = createSplits(job, job.jobId); // 确定Map Task数目numMapTasks:分片元数据信息数组的长度,即有多少分片就有多少numMapTasks job.numMapTasks = taskSplitMetaInfo.length; // 确定Reduce Task数目numReduceTasks,取作业参数mapreduce.job.reduces,参数未配置默认为0 job.numReduceTasks = job.conf.getInt(MRJobConfig.NUM_REDUCES, 0); // 省略部分代码 // 根据分片元数据信息计算输入长度inputLength,也就是作业大小 long inputLength = 0; for (int i = 0; i < job.numMapTasks; ++i) { inputLength += taskSplitMetaInfo[i].getInputDataLength(); } // 根据作业大小inputLength,调用作业的makeUberDecision()方法,决定作业运行模式是Uber模式还是Non-Uber模式 job.makeUberDecision(inputLength);由此,我们可以看出,作业运行方式Uber or Non-Uber是通过Job的makeUberDecision()方法,传入作业大小inputLength来确定的,本文,我们将研究这一话题,即如何确定作业运行方式Uber or Non-Uber?
在一文中我们了解了Uber和Non-Uber两种作业运行方式的含义,如下:
1、Uber模式:为降低小作业延迟而设计的一种模式,所有任务,不管是Map Task,还是Reduce Task,均在同一个Container中顺序执行,这个Container其实也是MRAppMaster所在Container;
2、Non-Uber模式:对于运行时间较长的大作业,先为Map Task申请资源,当Map Task运行完成数目达到一定比例后再为Reduce Task申请资源。
在确定了解上述内容后,我们再来看下Job的makeUberDecision()方法,这个Job的实现为JobImpl类,其makeUberDecision()方法代码如下:
/** * Decide whether job can be run in uber mode based on various criteria. * @param dataInputLength Total length for all splits */ private void makeUberDecision(long dataInputLength) { //FIXME: need new memory criterion for uber-decision (oops, too late here; // until AM-resizing supported, // must depend on job client to pass fat-slot needs) // these are no longer "system" settings, necessarily; user may override // 获取系统Uber模式下允许的最大Map任务数sysMaxMaps, // 取参数mapreduce.job.ubertask.maxmaps,参数未配置默认为9 int sysMaxMaps = conf.getInt(MRJobConfig.JOB_UBERTASK_MAXMAPS, 9); // 获取系统Uber模式下允许的最大Reduce任务数sysMaxReduces, // 取参数mapreduce.job.ubertask.maxreduces,参数未配置默认为1 int sysMaxReduces = conf.getInt(MRJobConfig.JOB_UBERTASK_MAXREDUCES, 1); // 获取系统Uber模式下允许的任务包含数据量最大字节数sysMaxBytes, // mapreduce.job.ubertask.maxbytes,参数未配置默认为远程作业提交路径remoteJobSubmitDir所在文件系统的默认数据块大小 long sysMaxBytes = conf.getLong(MRJobConfig.JOB_UBERTASK_MAXBYTES, fs.getDefaultBlockSize(this.remoteJobSubmitDir)); // FIXME: this is wrong; get FS from // [File?]InputFormat and default block size // from that // 获取系统为Uber模式设置的内存资源单元槽Slot大小sysMemSizeForUberSlot, // 取参数yarn.app.mapreduce.am.resource.mb,参数未配置默认为1536M long sysMemSizeForUberSlot = conf.getInt(MRJobConfig.MR_AM_VMEM_MB, MRJobConfig.DEFAULT_MR_AM_VMEM_MB); // 获取系统为Uber模式设置的CPU资源单元槽Slot大小sysCPUSizeForUberSlot, // 取参数yarn.app.mapreduce.am.resource.cpu-vcores,参数未配置默认为1 long sysCPUSizeForUberSlot = conf.getInt(MRJobConfig.MR_AM_CPU_VCORES, MRJobConfig.DEFAULT_MR_AM_CPU_VCORES); // 获取系统是否允许Uber模式标志位uberEnabled, // 取参数mapreduce.job.ubertask.enable,参数未配置默认为false,不启用 boolean uberEnabled = conf.getBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false); // 判断Map任务数是否满足系统为Uber模式设定的限制条件,结果赋值给smallNumMapTasks boolean smallNumMapTasks = (numMapTasks <= sysMaxMaps); // 判断Reduce任务数是否满足系统为Uber模式设定的限制条件,结果赋值给smallNumReduceTasks boolean smallNumReduceTasks = (numReduceTasks <= sysMaxReduces); // 判断任务包含数据量大小是否满足系统为Uber模式设定的限制条件,结果赋值给smallInput boolean smallInput = (dataInputLength <= sysMaxBytes); // ignoring overhead due to UberAM and statics as negligible here: // 获取系统配置的Map任务要求的内存大小requiredMapMB, // 取参数mapreduce.map.memory.mb,参数未配置默认为0 long requiredMapMB = conf.getLong(MRJobConfig.MAP_MEMORY_MB, 0); // 获取系统配置的Map任务要求的内存大小requiredReduceMB, // 取参数mapreduce.reduce.memory.mb,参数未配置默认为0 long requiredReduceMB = conf.getLong(MRJobConfig.REDUCE_MEMORY_MB, 0); // 计算要求的任务内存大小requiredMB, // 取Map任务要求的内存大小requiredMapMB与Reduce任务要求的内存大小requiredReduceMB中的较大者 long requiredMB = Math.max(requiredMapMB, requiredReduceMB); // 获取系统uber模式下Map任务要求的CPU核数requiredMapCores, // 取参数mapreduce.map.cpu.vcores,参数未配置默认为1 int requiredMapCores = conf.getInt( MRJobConfig.MAP_CPU_VCORES, MRJobConfig.DEFAULT_MAP_CPU_VCORES); // 获取系统uber模式下Reduce任务要求的CPU核数requiredReduceCores, // 取参数mapreduce.reduce.cpu.vcores,参数未配置默认为1 int requiredReduceCores = conf.getInt( MRJobConfig.REDUCE_CPU_VCORES, MRJobConfig.DEFAULT_REDUCE_CPU_VCORES); // 计算要求的任务需要CPU核数requiredCores, // 取Map任务要求的CPU核数requiredMapCores与Reduce任务要求的CPU核数requiredReduceCores中的较大者 int requiredCores = Math.max(requiredMapCores, requiredReduceCores); // 特殊处理:如果Reduce任务数目为0,即当为Map-Only任务时, // 要求的内存大小、CPU核数,以Map任务要求的为准 if (numReduceTasks == 0) { requiredMB = requiredMapMB; requiredCores = requiredMapCores; } // 当MR作业中任务要求的内存大小requiredMB小于等于系统为Uber模式设置的内存资源单元槽Slot大小sysMemSizeForUberSlot时, // 或者sysMemSizeForUberSlot被设定为不受限制时, // 确定为小内存要求,即标志位smallMemory为true boolean smallMemory = (requiredMB <= sysMemSizeForUberSlot) || (sysMemSizeForUberSlot == JobConf.DISABLED_MEMORY_LIMIT); // 当MR作业中任务要求的CPU核数requiredCores小于等于系统为Uber模式设置的CPU资源单元槽Slot大小sysCPUSizeForUberSlot时, // 确定为小CPU要求,即标志位smallCpu为true boolean smallCpu = requiredCores <= sysCPUSizeForUberSlot; // 确定作业是否为链式作业,并赋值给标志位notChainJob,ture表示非链式作业,false表示为链式作业 boolean notChainJob = !isChainJob(conf); // User has overall veto power over uberization, or user can modify // limits (overriding system settings and potentially shooting // themselves in the head). Note that ChainMapper/Reducer are // fundamentally incompatible with MR-1220; they employ a blocking // queue between the maps/reduces and thus require parallel execution, // while "uber-AM" (MR AM + LocalContainerLauncher) loops over tasks // and thus requires sequential execution. // 判断是否为Uber模式,赋值给isUber, // 判断的依据为,以下七个条件必须全部满足: // 1、参数mapreduce.job.ubertask.enable配置为true,即系统允许Uber模式; // 2、Map任务数满足系统为Uber模式设定的限制条件,即小于等于参数mapreduce.job.ubertask.maxmaps配置的值,如果参数未配置,则应该小于等于9; // 3、Reduce任务数满足系统为Uber模式设定的限制条件,即小于等于参数mapreduce.job.ubertask.maxreduces配置的值,如果参数未配置,则应该小于等于1; // 4、任务包含数据量大小满足系统为Uber模式设定的限制条件,即任务数据量小于等于参数mapreduce.job.ubertask.maxbytes配置的值,如果参数未配置,则应小于等于远程作业提交路径remoteJobSubmitDir所在文件系统的默认数据块大小; // 5、MR作业中任务要求的内存大小requiredMB小于等于系统为Uber模式设置的内存资源单元槽Slot大小sysMemSizeForUberSlot时,或者sysMemSizeForUberSlot被设定为不受限制; // 6、MR作业中任务要求的CPU核数requiredCores小于等于系统为Uber模式设置的CPU资源单元槽Slot大小sysCPUSizeForUberSlot; // 7、作业为非链式作业; isUber = uberEnabled && smallNumMapTasks && smallNumReduceTasks && smallInput && smallMemory && smallCpu && notChainJob; if (isUber) {// 当作业为Uber模式运行时,设置一些必要的参数 LOG.info("Uberizing job " + jobId + ": " + numMapTasks + "m+" + numReduceTasks + "r tasks (" + dataInputLength + " input bytes) will run sequentially on single node."); // make sure reduces are scheduled only after all map are completed // mapreduce.job.reduce.slowstart.completedmaps参数设置为1, // 即全部Map任务完成后才会为Reduce任务分配资源 conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 1.0f); // uber-subtask attempts all get launched on same node; if one fails, // probably should retry elsewhere, i.e., move entire uber-AM: ergo, // limit attempts to 1 (or at most 2? probably not...) // Map、Reduce任务的最大尝试次数均为1 conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 1); conf.setInt(MRJobConfig.REDUCE_MAX_ATTEMPTS, 1); // disable speculation // 禁用Map、Reduce任务的推测执行机制 conf.setBoolean(MRJobConfig.MAP_SPECULATIVE, false); conf.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, false); } else {// 当作业为Non-Uber模式时,通过info级别日志,输出作业不能被判定为Uber模式的原因,根据上述7个标志位判断即可 StringBuilder msg = new StringBuilder(); msg.append("Not uberizing ").append(jobId).append(" because:"); if (!uberEnabled) // Uber模式开关未打开,这种模式被禁用了 msg.append(" not enabled;"); if (!smallNumMapTasks) // 有太多的Map任务 msg.append(" too many maps;"); if (!smallNumReduceTasks) // 有太多的Reduce任务 msg.append(" too many reduces;"); if (!smallInput) // 有太大的输入 msg.append(" too much input;"); if (!smallCpu) // 需要占用过多的CPU msg.append(" too much CPU;"); if (!smallMemory) // 需要占用过多的内存 msg.append(" too much RAM;"); if (!notChainJob) // 是一个链式作业,无法使用Uber模式 msg.append(" chainjob;"); // 记录无法使用Uber模式的日志信息 LOG.info(msg.toString()); } }makeUberDecision()方法的逻辑十分清晰,但是涉及到的判断条件、参数比较多,总的来说,一个MapReduce是使用Uber模式还是Non-Uber模式运行,要综合考虑以下7个因素,这些条件缺一不可:
1、 参数mapreduce.job.ubertask.enable配置为true,即系统允许Uber模式,这是一个Uber模式的开关;
2、Map任务数满足系统为Uber模式设定的限制条件,即小于等于参数mapreduce.job.ubertask.maxmaps配置的值,如果参数未配置,则应该小于等于9;
3、Reduce任务数满足系统为Uber模式设定的限制条件,即小于等于参数mapreduce.job.ubertask.maxreduces配置的值,如果参数未配置,则应该小于等于1;
4、任务包含数据量大小满足系统为Uber模式设定的限制条件,即任务数据量小于等于参数mapreduce.job.ubertask.maxbytes配置的值,如果参数未配置,则应小于等于远程作业提交路径remoteJobSubmitDir所在文件系统的默认数据块大小;
5、MR作业中任务要求的内存大小requiredMB小于等于系统为Uber模式设置的内存资源单元槽Slot大小sysMemSizeForUberSlot时,或者sysMemSizeForUberSlot被设定为不受限制;
6、MR作业中任务要求的CPU核数requiredCores小于等于系统为Uber模式设置的CPU资源单元槽Slot大小sysCPUSizeForUberSlot;
7、作业为非链式作业。
前面6个条件在上面的描述和makeUberDecision()方法代码及其注释中都描述的很清晰,读者可自行查阅。
下面,我们重点看看第7个条件:作业为非链式作业,这个条件是如何判断的呢?它是通过isChainJob()方法来判断的,代码如下:
/** * ChainMapper and ChainReducer must execute in parallel, so they're not * compatible with uberization/LocalContainerLauncher (100% sequential). */ private boolean isChainJob(Configuration conf) { boolean isChainJob = false; try { // 获取取Map类名mapClassName,取参数mapreduce.job.map.class String mapClassName = conf.get(MRJobConfig.MAP_CLASS_ATTR); if (mapClassName != null) { // 通过Map类名mapClassName获取Map类Class实例mapClass Class mapClass = Class.forName(mapClassName); // 通过Class的isAssignableFrom()方法,看看mapClass是否为ChainMapper的子类,或者就是ChainMapper, // 是的话,我们认为它就是一个链式作业 if (ChainMapper.class.isAssignableFrom(mapClass)) isChainJob = true; } } catch (ClassNotFoundException cnfe) { // don't care; assume it's not derived from ChainMapper } catch (NoClassDefFoundError ignored) { } try { // 获取取Reduce类名reduceClassName,取参数mapreduce.job.reduce.class String reduceClassName = conf.get(MRJobConfig.REDUCE_CLASS_ATTR); if (reduceClassName != null) { // 通过Reduce类名reduceClassName获取Reduce类Class实例reduceClass Class reduceClass = Class.forName(reduceClassName); // 通过Class的isAssignableFrom()方法,看看reduceClass是否为ChainReducer的子类,或者就是ChainReducer, // 是的话,我们认为它就是一个链式作业 if (ChainReducer.class.isAssignableFrom(reduceClass)) isChainJob = true; } } catch (ClassNotFoundException cnfe) { // don't care; assume it's not derived from ChainReducer } catch (NoClassDefFoundError ignored) { } return isChainJob; }它实际上就是看Map或Reduce是否是ChainMapper或ChainReducer的直接或间接子类,或者就是二者,通过参数mapreduce.job.map.class、mapreduce.job.reduce.class取类名并利用Class.forName构造Class实例,然后通过Class的isAssignableFrom()方法判断Map或Reduce是否是ChainMapper或ChainReducer的直接或间接子类,或者就是二者,就是这么简单。
那么问题又来了,什么是链式作业?为什么继承了ChainMapper或ChainReducer就不能在Uber模式下运行?下面我们一一解答。
首先,链式作业是什么呢?有些时候,你会发现,一个单独的MapReduce Job无法实现你的业务需求,你需要更多的MapReduce Job来处理你的数据,而此时,将多个MapReduce Job串成一条链就形成一个更大的MapReduce Job,这就是链式作业。而链式作业实现的一个根本条件就是其Mapper或Reducer分别继承自ChainMapper和ChainReducer。
那么,为什么继承了ChainMapper或ChainReducer就不能在Uber模式下运行?连同什么是ChainMapper、ChainReducer这个问题,我们一起来做一个最直接最简单的解答,更多详细内容请查看关于专门介绍ChainMapper或ChainReducer的文章。
首先看下ChainMapper的实现,在其内部,有一个Chain类型的成员变量chain,定义并在setup()方法中初始化如下:
private Chain chain; protected void setup(Context context) { chain = new Chain(true); chain.setup(context.getConfiguration()); }而Chain中有两个最关键的变量,Mapper列表mappers和Thread列表threads如下:
private Listmappers = new ArrayList ();
private List在ChainMapper的run()方法内,会将Chain的mappers中每个Mapper通过chain的addMapper()方法添加至chain中,而chain的addMapper()方法本质上就是基于每个Mapper生成一个MapRunner线程,然后添加到threads列表内,然后再由Mapper启动chain中所有线程threads,关键代码如下:threads = new ArrayList ();
ChainMapper的run()方法
public void run(Context context) throws IOException, InterruptedException { setup(context); int numMappers = chain.getAllMappers().size(); if (numMappers == 0) { return; } ChainBlockingQueueChain的其中一个addMapper()方法> inputqueue; ChainBlockingQueue > outputqueue; if (numMappers == 1) { chain.runMapper(context, 0); } else { // add all the mappers with proper context // add first mapper outputqueue = chain.createBlockingQueue(); chain.addMapper(context, outputqueue, 0); // add other mappers for (int i = 1; i < numMappers - 1; i++) { inputqueue = outputqueue; outputqueue = chain.createBlockingQueue(); chain.addMapper(inputqueue, outputqueue, context, i); } // add last mapper chain.addMapper(outputqueue, context, numMappers - 1); } // start all threads chain.startAllThreads(); // wait for all threads chain.joinAllThreads(); }
/** * Add mapper that reads and writes from/to the queue */ @SuppressWarnings("unchecked") void addMapper(ChainBlockingQueue可以看出,ChainMapper实际上实现了一种多重Mapper,即multiple Mapper,它不再依托一个单独的Map Task,执行一种Map任务,而是依托多个Map Task,执行多种Map任务,所以,它肯定不适合Uber模式,因为Uber模式只限于Map、Reduce等各个任务的单线程串行执行。 ChainReducer也是如此,不再做特别的说明。> input, ChainBlockingQueue > output, TaskInputOutputContext context, int index) throws IOException, InterruptedException { Configuration conf = getConf(index); Class keyClass = conf.getClass(MAPPER_INPUT_KEY_CLASS, Object.class); Class valueClass = conf.getClass(MAPPER_INPUT_VALUE_CLASS, Object.class); Class keyOutClass = conf.getClass(MAPPER_OUTPUT_KEY_CLASS, Object.class); Class valueOutClass = conf.getClass(MAPPER_OUTPUT_VALUE_CLASS, Object.class); RecordReader rr = new ChainRecordReader(keyClass, valueClass, input, conf); RecordWriter rw = new ChainRecordWriter(keyOutClass, valueOutClass, output, conf); MapRunner runner = new MapRunner(mappers.get(index), createMapContext(rr, rw, context, getConf(index)), rr, rw); threads.add(runner); }
转载地址:http://lmzdx.baihongyu.com/