1. 从一条submit命令说起:Job提交链路的全景图我最早研究Hadoop的job提交是因为手头一个跑批任务频繁在提交阶段就失败,日志里翻来覆去就是各种莫名其妙的连接异常。当时被这个问题折磨了几天最后实在坐不住只能硬着头皮去啃源码。啃完之后才意识到以前对Hadoop的理解太黑盒了——网上铺天盖地的资料都在讲MapReduce怎么跑、怎么调优但很少有人把一条submit命令敲下去之后代码到底走了哪些分支、经过几次RPC、什么时候开始真正占用集群资源这条链路讲清楚。本文就围绕Hadoop job提交这条主线从源码层面逐层拆开看。适合三类读者一是被各种提交期报错折腾过的运维和平台开发二是刚接触YARN和MapReduce源码、想找一条清晰阅读路径的学习者三是需要二次开发客户端或自定义调度逻辑的进阶用户。我会尽量把涉及的类名、方法名、关键参数都列出来同时把每一步的为什么也讲透而不是只堆代码路径。在进入正文之前先把整条链路画个轮廓。一次标准作业提交大体走这么五站客户端初始化JobConf和Job对象用户代码在这里把jar包路径、主类名、输入输出路径、各类参数全部封装好。Job.waitForCompletion触发submitJob内部调用JobSubmitter提交器。JobSubmitter写分片、上传job资源jar、配置文件、分片元数据到HDFS临时目录。客户端通过RPC协议向ResourceManager提交Application并获取ApplicationId。RM创建并调度ApplicationMasterAM随后启动真正开始向RM申请容器执行Map和Reduce任务。后面所有源码分析都围绕这五站展开。先声明一下本文所有代码路径基于社区版2.x/3.x仍通用的核心流程个别细节在不同小版本里会有微调但不影响主线理解。想快速上手的读者建议本地拉一份源码用IDE跟着文章逐步打断点效果比只看文字好得多。2. 客户端侧的初始化与Job对象构建:脏活累活都在这里2.1 JobConf、Job和Configuration三者的关系很多初学者一上来就被三个类搞蒙JobConf、Job、Configuration。其实这三者的分工非常明确Configuration是通用的配置存储容器本质就是一个巨大的XML/properties键值对集合JobConf继承自Configuration专门为MapReduce作业补充了setInputFormat、setOutputFormat这一组类型安全的方法Job则是一个更上层的门面类把提交、监控、获取结果这些操作全部封装成对外API。在提交链路里JobConf负责把用户设置的Map数量、Reduce数量、各类路径、压缩配置固化下来Job在构造时会把JobConf引用保存到内部后续提交时直接从这份配置快照中读取。2.2 提交前用户代码到底做了什么以最常见的代码写法为例Configuration conf new Configuration(); Job job Job.getInstance(conf, MyJob); job.setJarByClass(MyMapper.class); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1);这里有一个非常容易忽略的细节Job.getInstance(conf, MyJob)这一步内部其实就会构造Job并生成唯一的JobID只是这个ID在submit之前是临时版。直到真正提交到RM客户端才会拿到RM分配的正式ApplicationId再用新ID重建Job对象。源码里对应这段逻辑:public Job(Configuration conf, String jobName) throws IOException { this.conf new JobConf(conf); if (this.conf.get(mapreduce.job.name) null) { this.conf.set(mapreduce.job.name, jobName); } this.jobId null; // 此时还没有真正的ID }所有的setter方法本质都是在给this.conf这个JobConf塞键值对。后面提交阶段会有一次配置规范化过程把用户写的mapreduce.*前缀参数统一转换并写入job.xml。2.3 waitForCompletion的阻塞与内部逻辑waitForCompletion做的事情看起来简单实际上承担了提交、轮询、日志下载三合一的功能。源码精简后如下public boolean waitForCompletion(boolean verbose) throws IOException, InterruptedException, ClassNotFoundException { if (state JobState.DEFINE) { submit(); } if (verbose) { monitorAndPrintJob(); } else { // 直接轮询状态 } return isSuccessful(); }重点在submit()方法里它会先判断当前job有没有cluster对象如果没有就通过JobConf里的mapreduce.framework.name来决定使用哪种提交器。默认是yarn对应YARNRunner老版本可能会走到local runner也就是本地模式。这一处判断也是很多初学者排错时容易忽略的分水岭部署在集群上但配置文件里framework.name被改成了local结果提交走了假流程作业秒完成输出目录空空如也。3. JobSubmitter运行期准备:分片计算与资源上传的幕后逻辑3.1 写分片文件时那些容易被忽略的边界条件真正进入提交逻辑后第一个重量级动作是写分片文件。JobSubmitter内部的writeSplits方法负责调用InputFormat.getSplits计算分片然后序列化写入临时目录下的 splits 文件。这里有几个关键点值得展开分片数量不直接等于Map任务数而是split的个数。每个split最终会对应一个map task但中间可能经过CombineFileInputFormat的合并或自定义InputFormat的改写。getSplits的返回值实际上是一个List 每个InputSplit的接口包含getLength和getLocations两个方法前者决定数据倾斜计算时的权重后者决定任务本地性调度时参考的节点列表。FileSplit的存储路径、起始偏移和长度这三个字段会直接影响数据读取的准确性因此源码里对这个结构做了非常严谨的序列化设计。在使用FileInputFormat的情况下有一个参数经常被忽略mapreduce.input.fileinputformat.split.minsize和maxsize。默认情况下split大小等于block大小但如果你设置了maxsize小于block size一个block会被切分成多个split此时单个split未必包含完整块读取数据时需要通过FSDataInputStream的seek定位真实偏移。这个机制在归档文件或小文件合并场景下特别有用。3.2 资源上传的临时目录结构与文件命名规则分片算完后JobSubmitter会调用uploadFiles方法把job资源上传到HDFS上的临时提交目录。这里的临时目录路径格式我记了好几年几乎每个排查找不到作业资源的case都会用到${mapreduce.cluster.local.dir}/job_${timestamp}_${id}/libjars ${mapreduce.cluster.local.dir}/job_${timestamp}_${id}/files ${mapreduce.cluster.local.dir}/job_${timestamp}_${id}/job.xml ${mapreduce.cluster.local.dir}/job_${timestamp}_${id}/split其中mapreduce.cluster.local.dir对应HDFS上的staging目录默认是/tmp/hadoop-yarn/staging/用户名/.staging。这里有个坑如果用户无权限写这个目录或者目录空间不足提交会直接失败。生产环境建议为每个业务账号单独指定staging目录并做容量规划而不是让所有作业都往默认目录挤。job.xml是整个作业配置的序列化快照包含jobConf里所有参数的明文键值对因此如果有敏感配置属性不要写在作业配置里而是放在受控的配置文件里单独分发。这是很多企业安全扫描会盯的点。3.3 上传前的凭据与token处理细节在Kerberos环境下作业提交还会涉及token的传递。JobSubmitter构造时会通过Credentials对象保存当前用户的tokens并在上传的job.xml旁边生成一个.jobToken文件后续AM启动时会从这个文件里读取凭证。源码中对应的位置是JobSubmitter.writeConfigFile之前对credentials的复制操作。很多线上报错如File does not exist: /tmp/.../staging/xx/.jobToken都是因为token文件缺失或权限不对本质上是提交用户和AM运行用户不一致导致的。这一块的排查思路一般是三步先确认当前提交用户是否有staging目录的写权限再看提交和运行用户是否一致最后检查token文件是否存在以及过期时间。三个都没问题基本不会再碰到诡异的认证错误。4. 客户端与ResourceManager的RPC握手:ApplicationId是这样来的4.1 提交用的协议接口和线程模型当所有准备动作完成JobSubmitter会调用YARNRunner的submitJob最终要走ClientRMProtocol的RPC接口。这个接口的提交方法签名是submitApplication(SubmitApplicationRequest)客户端侧对应的代理类是ApplicationClientProtocolPBClientImpl。RPC通信走的还是Hadoop自身的IPC框架也就是说序列化用的是Writable机制传输层用Netty或Socket。客户端会先通过YarnConfiguration从yarn.resourcemanager.address解析RM的主机名和端口建立连接后完成一次握手包括版本校验和用户认证。版本不匹配是这里最常见的错误客户端hadoop版本和RM版本差得太多直接抛HadoopIllegalArgumentException。对高可用集群来说这里的解析逻辑会更复杂——客户端需要先访问RM的leader选举状态通过ActiveYarnRM获取当前active RM的地址。这个逻辑在源码中由YARNHAUtil处理涉及ZooKeeper的连接和leader信息的读取。很多开发者在本地模拟HA环境时往往忽略zookeeper集群的启动顺序导致RM明明起来了但客户端就是解析不到地址。4.2 RM侧对提交请求的处理管线SubmitApplication请求到达RM后会先经过RMAppManager的submitApplication方法。这里做的第一件事不是开容器而是把Application信息登记到RM的state store中。RM内部会生成RMAppImpl实例并将app状态机推进到SUBMITTED状态随后通过Dispatcher异步触发ApplicationScheduler的调度流程。稍微展开一下状态机RMAppImpl的状态包括NEW: 刚创建尚未接收任何事件NEW_SAVING: 正在持久化到state storeSUBMITTED: 已持久化完成准备提交给调度器ACCEPTED: 调度器接受并分配了AM containerRUNNING: AM已启动并注册成功FINISHED / FAILED / KILLED: 终态大部分客户端看到Application has been submitted successfully后卡住不动往往是RM的app状态从SUBMITTED往ACCEPTED转换失败——要么是队列资源不足要么是调度器配置了ACL拒绝。此时去RM日志里搜Unable to add application或者Application is not ACCEPTED一般能直接定位。4.3 从临时JobID到正式ApplicationId的转变在客户端提交成功后返回的SubmitApplicationResponse里携带了ApplicationId客户端用这个ID重新构造Job对象并创建对应的JobStatus之后的所有状态查询、日志获取、计数器读取都通过这个正式ID进行。这里有一个容易踩的坑由于客户端在提交前生成的临时ID和提交后拿到的正式ID不同如果在用户代码里提前把临时ID打印出来会看到和ResourceManager UI上完全不一样的编号容易让人误以为提交了两次。实际这是正常现象不是bug。5. ApplicationMaster的启动与资源分配:作业真正开始跑起来的地方5.1 AM container从分配到启动的全过程RM的调度器接受应用后会为AM分配一个container。这个container资源量默认为1GB内存和1个vcore对应的配置项是yarn.app.mapreduce.am.resource.mb和yarn.app.mapreduce.am.resource.cpu-vcores。有很多开发者在提交大作业时只调高了map和reduce的资源配置忽略了AM自身资源结果AM频繁被NodeManager杀容器作业一直处于重启循环。诊断方法很直接去AM的日志目录看是否频繁出现Container killed by the NodeManager或OutOfMemory。AM容器启动后内部第一个运行的是MRAppMaster的主类。它启动时会搭建一系列服务包括ContainerLauncher: 负责与NodeManager通信启停容器TaskAttemptListener: 接收任务完成的心跳反馈JobHistoryEventHandler: 写入job history文件MapTaskCompletionEventsUpdateThread: 处理mapper完成事件向reducer的传递这些服务全部初始化完毕后MRAppMaster会调用RM的AMRMProtocol注册自己然后周期性地向RM发送心跳申请资源。5.2 容器申请与优先级:一次典型的资源请求分析AM内部维护了自己的一套任务调度逻辑。它会根据分片的个数和位置信息为每个map task生成一个ContainerRequest。这个请求里最关键的两个字段资源量Resource和节点亲和性NodeId列表。由于map task的本地性直接影响数据读取效率AM会优先让task落到HDFS块所在的节点上。源码里对应的是TaskScheduler的assignTasks方法。它内部维护三个队列psuedoDistributedMode: 单节点仿真模式正常模式下的事件队列和任务尝试队列assignTasks会根据当前可用资源和task需要的资源进行匹配分配成功的task会被包装成ContainerLaunchContext交给ContainerLauncher异步启动。这里有个小技巧如果你希望map task尽量本地化可以调整参数mapreduce.job.reduce.slowstart.completedmaps这个参数决定reduce task在多少比例的map完成后才启动默认0.05也就是说map跑5%后reduce就开始申请资源如果设置成1.0则reduce会等所有map完成才启动这能显著减少reduce等待map输出的时间但也会延长整体作业时长。5.3 从AM视角看任务的失败与重试任务失败后的重试逻辑也在AM内部完成。JobImpl中的EventHandler处理task attempt的失败事件并根据重试次数决定是再次调度还是将整个job标记为失败。默认每个map task重试4次reduce task重试4次参数是mapreduce.map.maxattempts和mapreduce.reduce.maxattempts。实际排错时最容易被忽略的是失败原因是否真的在task本身。很多情况下task OOM或者磁盘写满会被误判为代码问题实际上是容器资源不足。YARN会自动把这种容器失败反馈给AMAM再决定是否重试。一个常见优化是把mapreduce.map.memory.mb从默认的1GB调高到2GB同时配套调高mapreduce.map.java.opts的堆大小。这两个参数不匹配会导致JVM堆申请超过容器内存限制task直接被NM杀掉。6. 提交链路上的经典故障模式与排查方法论6.1 提交卡住不动的两类根因与定位手段作业提交后如果长时间没有进入RUNNING状态先不要急着怀疑代码。从客户端日志和RM日志交叉定位通常两类根因资源问题: 目标队列资源不足或超额申请被拒绝。看RM UI的Scheduler页面观察队列的Pending资源是否一直大于0同时检查队列ACL是否限制了该用户提交。认证和授权问题: Kerberos票据过期或token缺失。这类问题通常伴随明显的Kerberos异常堆栈直接搜客户端日志中的GSSException或AuthorizationException即可。6.2 分片文件为空或路径找不到的排查链路提交阶段报Input path does not exist和splits file empty是两个常被误判的错误。前者的根因往往是输入路径写的是相对路径而HDFS工作目录与客户端本地工作目录不同。后者的根因通常是无匹配文件或InputFormat的isSplitable返回false导致单个split覆盖整个目录但内部无数据。顺带说一句isSplitable的返回值还会影响是否走combine机制这个参数在做小文件优化时也十分关键。6.3 诊断RPC通信层故障的常用工具与命令遇到RPC连接问题我喜欢先用hadoop自带的调试手段确认网络可达性和版本兼容性hdfs dfs -ls /tmp # 先确认HDFS端口和认证正常 hadoop jar myjob.jar MyJob input output -D mapreduce.framework.nameyarn -D yarn.resourcemanager.addressrm-host:8032如果客户端日志里有Connection refused八成是端口配错或RM进程没起来。用telnet验证端口再检查yarn-site.xml中yarn.resourcemanager.address是否配置了服务地址而非候选地址列表。这里在HA环境下有个常见坑yarn.resourcemanager.address如果被设置成logical name而不是active节点的物理地址客户端是无法解析的需要依赖yarn.client.failover-proxy-provider配置。6.4 那些被反复踩的提交期坑位汇总现象根因快速处理提交后秒失败无RM日志本地模式生效framework.name被覆盖确认mapreduce.framework.nameyarnApplication卡在ACCEPTEDAM资源不足或队列ACL限制调大AM资源检查队列权限找不到job.xmlstaging目录权限不足或被清理检查/tmp下staging目录换指定目录客户端和RM版本不一致IPC协议不兼容统一客户端和集群版本大版本对齐task被反复kill容器内存与实际JVM堆不匹配同步调整memory.mb和java.opts本地性极差分片location信息丢失检查FileInputFormat的split计算日志7. 提交参数调优的实战经验:让提交不再成为瓶颈7.1 从一次真实调优案例看参数组合效果某次数据处理任务mapper数量3000多个AM默认1GB资源跑到一半AM日志频繁出现重试。当时我用了三组参数调整-D mapreduce.job.reduces100 -D yarn.app.mapreduce.am.resource.mb2048 -D yarn.app.mapreduce.am.command-opts-Xmx1536m第一组减少reduce数量降低AM调度压力第二组和第三组保证AM自身有充足内存。调整后作业稳定跑完提交到进入RUNNING的时间从4分钟缩短到40秒。这个案例说明很多时候不是代码慢而是AM和客户端之间的资源博弈没调好。7.2 资源参数之间的联动关系与取值技巧AM、map、reduce三者的资源参数互相影响调整时必须联动。核心原则JVM堆参数需要小于等于容器总内存的75%左右留出部分空间给非堆内存和系统开销。有一套简单公式可以套用mapreduce.map.memory.mb mapreduce.map.java.opts堆大小 256MB以上余量reduce同理AM参数同理有个参数特别容易被忽略yarn.nodemanager.resource.memory-mb 是NodeManager分配给所有容器总内存的硬上限。如果你给单个容器设得很大但NM总资源不够同样会调度失败。统计每个节点可用内存后合理规划单容器大小才能让集群不被浪费。7.3 常见JVM参数与容器资源对照速查表容器内存(mb)建议JVM堆(Xmx)适用场景1024768m默认轻量级任务20481536m中等数据量ETL40963072m复杂join或大聚合81926144m大规模机器学习特征处理8. 扩展视野:从提交源码反推二次开发的可能性读完了提交链路的源码很容易发现一些可扩展的地方。比如自定义ResourceManager的调度器或者为特定业务定制ContainerRequest的优先级策略都需要精确掌握这条提交链路的时序。再比如想在提交阶段对作业配置做校验或改写可以在JobSubmitter上传前拦截job.xml做动态参数注入。业内一些平台型工具本质上就是在这条链路上做了大量的自动化和策略化处理。对于已经稳定运行的集群即使不做二次开发阅读源码也能带来一个直接好处很多日志里的英文缩写和状态码你不再需要去全网搜因为你知道它们是在哪个方法里产生的意味着什么。这种知其然也知其所以然的能力是排查线上问题最值钱的资本。如果你准备深入这块我的建议是找一条最简单的WordCount从Job.waitForCompletion开始打断点一路跟到MRAppMaster的run方法。整个过程不用太久但跟完之后你对分布式计算框架的理解会和只看文档完全不一样。最后再分享一个命令行层面的小技巧提交作业时加上-D mapreduce.job.debug.verbosetrue客户端会在提交阶段打印所有动态生成的配置项和分片信息。这个参数在对比不同版本集群行为差异时极其好用尤其是排查同一个jar包在不同环境行为不一致的问题时一开就能看出配置差异。