★ FEATURED ARTICLE 先解答标题里的那个问题输入分片到底由谁决定答案是客户端不是NameNode。NameNode在整个过程中只做一件事——把文件切成了哪些块、每块在哪些节点上的数据报给客户端剩下怎么拼、怎么切、切多大、map任务的对齐策略全部由提交作业的客户端代码算出来。这个结论听起来简单实际运行中却埋着不少坑。我见过不少同学以为调了分区数/并行度就能控制map数量结果作业跑起来完全不是那回事回去改配置、重启、再观察来回折腾半天。这篇文章会把“客户端决定分片”这条链路从头到尾讲透涉及InputSplit的本质、客户端在作业提交阶段做了什么、NameNode提供了哪些信息、为什么这个设计能撑住大规模集群以及日常调优和故障排查中我们到底应该盯住哪些参数和日志。无论你是刚接触MapReduce还是已经在维护一批批处理作业弄清楚这条链路都会少走很多弯路。1. 先对齐概念输入分片和三个基础角色1.1 输入分片到底是什么输入分片InputSplit是一个很底层的概念先明确下来。HDFS上文件会被物理切成若干块block每个块默认大小是128MB或256MB分别存放在不同节点上。但MapReduce处理的单位不是“块”而是“逻辑分片”。一个分片描述一段数据范围比如“文件A的第0到128MB”“文件A的128MB到256MB”它不一定和物理块完全对齐也可以跨多个物理块。分片和map任务是一一对应的一个分片最终由一个map任务处理。所以map数量并不是你随便设的它基本等于输入数据被切出来的分片数量。你看到的“map任务数不符合预期”“这个作业为什么起了这么多map”本质上是分片计算出现了你没预料到的结果。用生活化的方式来理解HDFS把一个大文件看成一整箱货物分块等于把货物拆到不同的货架上而输入分片则是运输清单上的一次搬运任务可以搬半箱、也可以搬一箱半搬运任务的数量决定了你要派多少辆车。这个“运输清单”在谁手里在客户端手上。1.2 客户端、NameNode、DataNode各管哪一段搞清楚这三个角色的分工才能明白“为什么是客户端而不是NameNode”这个问题的前提。角色负责什么接触业务数据吗参与分片计算吗客户端提交作业、解析参数、计算分片、生成作业配置不直接读取数据但计算分片会发起元数据请求是分片计算的主要执行者NameNode维护文件系统的目录树、文件块映射、块所在节点列表不接触业务数据只管理元数据否只提供原始信息DataNode实际存储数据块处理读写请求是业务数据在这里否只被动响应读写重点看NameNode它本质上是一个元数据服务维护“哪个文件由哪些块组成每个块被复制到哪几个节点”。它非常清楚每个文件的大小、每个块的位置但它不会替你决定“这个文件应该切成几个逻辑分片、每个分片包含哪几个块”。这些计算逻辑在客户端代码里。这就像你去图书馆借一整排书管理员能告诉你每本书在哪个书架、哪个位置但他不会替你想好“你一次搬几本、分几趟搬”。怎么搬是你自己的规划问题。1.3 为什么现在还要搞清楚这条链路有人可能会想现在是Spark、Flink的时代MapReduce我都快不写了还有必要深究InputSplit吗有而且很有必要。Spark从HDFS读数据时也有分区partition概念底层同样会用到Hadoop的InputFormat或者遵循类似的split划分思想。Hive跑MR引擎时每个Map Task的数量也是由分片决定。你在Hive里执行一条SELECT COUNT(*)如果底层是Tez或MR文件怎么切、Map数量是多少照样受这些逻辑影响。可以说只要数据源是HDFS、计算框架还依赖Hadoop生态的输入体系分片决策这条链路就绕不开。2. 一次分片决策的完整链路客户端主导NameNode只是“资料员”2.1 作业提交时客户端在干什么写过一个MapReduce作业的人都知道提交作业的入口是JobClient或JobSubmitter。在真正把任务交给YARN之前客户端会先在本地完成一堆准备工作其中最关键的一步就是写入分片信息。这里很多人会误会以为是YARN的ResourceManager收到作业后再去做分片规划。不是这样的。作业还没提交到ResourceManager之前客户端就已经把分片算好了并且把结果序列化成一个文件跟作业配置一起提交到集群暂存目录。ResourceManager和ApplicationMaster只是拿到这个已经算好的清单按清单去分配资源。具体流程大致是客户端解析作业参数确定输入路径。调用InputFormat.getSplits(JobConf)方法。FileInputFormat及其子类向FileSystem发起请求获取输入路径下所有文件的状态。对每个文件通过getFileBlockLocations向NameNode询问文件的块信息包括每个块的偏移量、长度和所在节点列表。客户端根据块位置信息、分片大小上限、下限逐文件算出分片列表。把分片列表序列化到job.split文件并生成job.splitmetainfo一起写入staging目录。提交作业给ResourceManagerApplicationMaster启动后会去读job.split文件按其描述为每个分片申请container并启动map任务。注意第4步和第5步NameNode接到的是“给我看看这个文件的块列表”这种查询请求而不是“请你帮我切一下这个文件”。真正执行切割算法的是客户端JVM里的FileInputFormat代码。2.2 getSplits的核心逻辑默认情况下FileInputFormat计算分片的核心逻辑可以精简成下面这几步拿到文件的长度length。根据配置计算分片大小math.max(minSize, math.min(maxSize, blockSize))。默认情况下minSize是1maxSize是很大一个数所以分片大小默认等于blockSize。minSize对应参数mapreduce.input.fileinputformat.split.minsize。maxSize对应参数mapreduce.input.fileinputformat.split.maxsize。从文件第一个字节开始以分片大小为步长切分直到文件末尾。每个分片会记录文件的路径、起始偏移、长度以及一个“优先节点列表”。如果文件不满一个分片大小那它就是一个独立分片。如果文件是0字节默认还会生成一个空分片确保map任务至少有一个。举例说明一个2.6GB的文件块大小128MB。如果不改任何参数分片大小就是128MB。那么2.6GB大约会被切成21个分片因为2.6GB除以128MB约等于20.8向上取整是21。这意味着这个作业的map数量大约是21不是20也不是你随便在代码里写的某个值。有人会问分片大小必须和块大小一样吗并不是。你如果把split.maxsize调到64MB那同一个文件会被切成更多更小的分片map任务数翻倍如果你把split.minsize调到256MB分片就会大于块一个分片可能包含多个物理块。这种情况通常出现在你想让Map Task变大、减少Map数量、或者避免过多小任务调度开销时。2.3 分片信息落在哪里、后面怎么用分片算完之后需要落到文件系统上否则ApplicationMaster无法知道该启动多少个map。客户端会把分片信息写到自己在HDFS上的staging目录通常路径类似/tmp/hadoop-yarn/staging/user/.staging/jobId/核心文件是job.split。ApplicationMaster启动后会读取job.split文件反序列化出InputSplit[]数组。每个分片包含什么信息主要就是文件路径哪个文件起始偏移从哪个字节开始长度这段数据有多长locations优先在哪几个节点上启动任务到了这一步YARN才参与进来ApplicationMaster为每个split申请一个container。因为split里带了节点列表调度器可以优先把container放到离数据最近的节点上这就是“数据本地性”data locality的由来。所以整个链路串起来是这样的客户端计算分片 - 分片信息持久化 - AM读取分片 - AM申请资源 - NodeManager启动map任务。NameNode从头到尾只扮演了资料提供者提供文件块的位置信息真正的决策点在客户端。3. 为什么这个决定落在客户端而不是NameNode3.1 NameNode的核心职责不允许它做分片要理解这个设计先要理解NameNode的定位。NameNode是整个HDFS的“大脑”任何文件操作都绕不开它创建文件、删目录、读块位置、写块确认……在大集群里NameNode每秒处理的元数据请求量非常巨大它的CPU和内存带宽是极其宝贵的资源。如果分片计算也放到NameNode上会出现什么情况每次作业提交NameNode不仅要响应“返回这个文件的块列表”还要额外承担“按特定参数把这个文件切成多少片、每片落在哪些节点”的计算任务。不同作业、不同租户的分片参数千差万别今天有人把split.maxsize调成200MB明天有人把minsize调成256MBNameNode就得为每次请求执行一遍不同的计算逻辑。问题还不仅仅是CPU开销。为了完成分片计算NameNode要维持每个文件更多的工作状态比如文件当前正在被几个作业切片、分片结果缓存不缓存、缓存失效怎么处理……这会把NameNode从“元数据服务”变成“作业规划服务”。它本来就承担着集群可用性的单点责任再把作业切片的复杂度缠进去整个集群的稳定性和扩展性都会受到挑战。日志和时间线也很容易陷入混乱。具备大规模集群生产经验的朋友都清楚“作业到底在谁那里生成计划”这个问题非常关键。定位慢作业时我们希望分片计算发生在客户端JVM里这样客户端日志就是第一手线索如果挪到NameNode端崩溃了都不知道是哪个作业引起的。3.2 客户端做分片的三个现实优势从工程实现角度把分片决策权放在客户端有很直接的好处。第一可插拔性强。InputFormat是接口不同业务的输入可以完全不同普通文件用TextInputFormat小文件多可以用CombineFileInputFormat数据库读取有DBInputFormat。这些实现全部在客户端加载不需要改动服务端代码。如果把分片逻辑固定死在NameNode上那么想扩展一种新的分片策略就得升级整个NameNode风险高、周期长。第二作业级别隔离。分片参数是每个作业自己的配置客户端计算时完全按照自己的配置来。比如同一个文件A作业想按128MB切片B作业想合并小文件后按256MB切片二者互不影响。这种“各算各的”模式天然适合多租户共享一个HDFS的场景。第三容错边界清晰。客户端算分片算到一半失败了只会影响当前作业的提交过程不会波及集群上的其他任务。NameNode即使感知到客户端请求超时也只是返回一个超时错误而已。故障影响范围完全可控。3.3 数据本地性到底是怎么锚定的分片计算里最容易忽略的是“locations”字段。客户端在询问NameNode某个文件的块位置时NameNode不仅会返回块大小还会返回这个块所在节点的列表而且列表是按机架感知排序的。比如一个文件块有三个副本位置分别是node-a、node-b、node-c那么三个节点都会出现在列表里。客户端的FileInputFormat在构造每个分片时会收集所有相关块的节点信息取一个交集这个交集就是该分片的“优先位置列表”。如果一个分片恰好对齐一个物理块那么它的优先位置就是该块所在的几个副本节点如果一个分片横跨多个块那么优先位置就会取多个块节点的交集交集为空时优先位置会退回成所有块节点的并集。这个列表在后续YARN调度时非常重要。ApplicationMaster为split申请container时specified这些节点列表调度器尽量把container放到列表中的节点上从而让map任务读取本地数据而不走网络。可以这么说NameNode提供的是“数据在哪”的事实客户端把事实加工成“任务应该放哪”的建议调度器拿着建议去执行。三步协作各司其职。4. 调分片前先学会看这三件事4.1 想调Map数量先搞清楚对应哪个参数不少同学调Map数量的方式是“感觉Map少了就调大”这属于没有看穿分片机制。分片大小由三个值共同决定split.minsize、split.maxsize、blockSize最终取max(minsize, min(maxsize, blockSize))。配置默认值作用mapreduce.input.fileinputformat.split.minsize1分片大小下限想让分片变大就调这个mapreduce.input.fileinputformat.split.maxsizeLong.MAX_VALUE分片大小上限想让分片变小、map变多就调这个dfs.blocksize128MB或256MBHDFS物理块大小默认决定分片基准几个实操口诀想让map任务变少把split.minsize调大。比如全部输入是500MB块大小128MB默认会切4到5个分片你把minsize调到256MB分片会变成2个或3个。想让map任务变多把split.maxsize调小。比如你希望每个Map只处理64MB就把maxsize设成64MB。改完配置后客户端日志会有类似number of splits:21的输出先看这个数字是否符合预期再决定要不要继续调。一句话总结你调的不是“Map数量”你调的是“分片大小”Map数量只是分片大小的最终结果。4.2 小文件场景是个例外默认的FileInputFormat有个让人头疼的缺点每个文件至少产生一个分片。也就是说如果输入目录下有10万个1KB的小文件即使总数据量只有100MBmap任务也会被切出10万个。这种“一文件一分片”的行为对小文件非常不友好。生产上处理小文件通常有两条路。一是数据源头治理用小文件合并工具把文件合并成大文件二是在读取阶段使用CombineFileInputFormat。这个类会把多个小文件合并到一个分片里合并时会考虑节点分布和文件大小尽可能把同一节点上的小文件凑在一起避免map数量爆炸。如果你在用Hive也可以关注Hive的hive.input.format配置换成CombineHiveInputFormat后小文件的Map任务数量会被显著压缩。这些方案本质上都是替换了“计算分片的客户端代码”而不是去改NameNode的行为。4.3 分片算得没错但任务还是在远程读数据数据本地性是分片机制里最容易出问题的环节。常见的一种情况一个4GB的文件默认128MB分片理论上每个split正好对应一个块map任务应该本地读。但如果你打开YARN的日志发现大量map任务的type不是NODE_LOCAL而是RACK_LOCAL甚至OFF_SWITCH说明container没有被放到该放的地方。原因可能出在几个方面集群负载高数据本地节点没有空闲container。这属于资源竞争修改优先级参数或者提高mapreduce.job.reduce.slowstart.completedmaps都不能直接解决本质是容量问题。文件块副本数只有1某个节点宕机了所有分片都只能远程读。一个split横跨两个块两个块的节点集合没有交集优先级列表变成了并集导致任务被随机分配。排查思路很简单打开作业的Counter查看“Data-local map tasks”、“Rack-local map tasks”两个指标。如果本地率长期低于70%就要考虑是不是资源不足或者副本策略不合理。4.4 自定义InputFormat的正确姿势当你确定默认的分片策略满足不了业务需要自定义时核心实现其实不复杂继承FileInputFormat重写createRecordReader和getSplits方法。注意getSplits不是一定要重写如果你的目标是“按行数切分”“按自定义边界切分”就必须在这里处理。我见过一个比较典型的场景某个离线系统按业务字段对齐文件里每条记录跨多个物理块默认按块切分会把同一条记录拆到两个map里导致后续处理需要额外的二次聚合。他们的做法是重写getSplits让分片边界对齐业务分隔符同时结合NameNode返回的块位置计算出每个分片的节点列表。自定义InputFormat的调试建议是先在小集群上打印每个split的路径、offset、length和locations确认和预期一致后再放到生产作业里。因为在客户端JVM里就可以直接输出分片清单调试其实很轻松不需要反复提交作业去观测。5. 实战问题速查表与我的经验5.1 高频问题与排查思路把平时运维中遇到的分片相关高频问题整理成一张表收藏起来能少踩不少坑。现象原因处理方式Map数量比预期多很多输入目录有大量小文件默认一文件一分片合并文件或使用CombineFileInputFormatMap数量比预期少split.maxsize被调大或文件本身就是小于blockSize的少量文件检查相关配置文件确认分片大小修改了split.minsize但map数量没变文件本身小于minsize或只有1个文件且块大小为上限确认输入文件实际大小看是否被上限约束map任务大量出现RACK_LOCAL集群资源紧张或副本数过少检查数据本地率增加副本数或错峰调度作业卡在“Running job”不启动mapjob.split文件生成失败或AM读取异常查看客户端staging目录检查job.split是否正常同样的文件不同作业map数量差异大各作业的split配置或InputFormat实现不同对比两批作业的InputFormat参数设置设置了并行度但map数完全没变MapReducemap数量由分片决定并行度是另一回事区分map数量与reduce/任务并发度用正确参数控制5.2 排障时我习惯看的第一手信息每次遇到“Map数量不对劲”我不会急着改参数而是先看三样东西第一样输入目录的文件列表。用hadoop fs -ls看一下文件数量、每个文件大小。如果文件数量上百且每个都很小问题几乎可以锁定为小文件导致的分片过多。第二样作业提交日志中的分片数。客户端日志或应用时间线里会有number of splits:之类的信息看到这个数字后手算一遍总大小除以期望的split大小是否对得上。第三样作业Counter中的数据本地率。打开MapReduce任务页面的Counter表格找到Data-local和Rack-local的统计。本地率低时优先看集群容量和资源调度不要先怀疑分片算法。这三样信息看下来80%的分片问题都能直接定位。剩下的20%基本是网络文件系统、加密区、或是自定义InputFormat的边界逻辑需要结合具体业务代码进一步看。5.3 我踩过的一个典型坑分享一个自己踩过几次的教训。有一次处理一批日志文件文件总大小约20GB单文件大小不一最大值约3GB。当时任务跑得很慢map任务数高达400多个。我第一反应是调大split.maxsize想减少map数量结果改完重启map数量只少了一点点并没有达到预期。后来看了文件列表才发现这批文件里有几百个几百KB的小文件。它们每个都独立切成一个分片占掉了大量map名额。真正的问题是文件碎片化不是split size设置不合理。后来改用CombineFileInputFormat同时把这些小文件在数据入库阶段做了合并map数量立刻降到几十个作业整体耗时下降了将近一半。这个案例说明一个道理分片参数只是杠杆数据形态才是根源。调参数调的是“杠杆怎么放”但如果不先看清数据文件本身长什么样杠杆放得再准也解决不了问题。6. 一句话记住客户端和NameNode的关系如果你只想从这篇文章里带走一句话那应该是分片是客户端的本地计算结果NameNode只是提供文件块位置和大小清单的“档案管理员”。两者在作业提交阶段短暂交互过一次之后分片就以文件形式交给YARN去执行。既然提到了“为什么”这里再补充一个设计层面的体会。把分片决策放在客户端表面上损失了“集中统一规划”的能力但换来了极大的灵活性和故障隔离能力。在一个大规模共享集群里“决策分散在边缘、资源集中由核心管理”是非常合理的架构它让最核心的NameNode和ResourceManager保持轻量、快速、稳定同时把和业务紧密相关的计算逻辑开放给客户端自由扩展。我个人的经验是排查分片问题不要一上来就去翻NameNode的日志。NameNode日志里通常只会出现类似于getBlockLocations的元数据请求根本不会反映分片计算过程。分片问题一定先看客户端日志、输入文件列表和YARN的调度信息。把这几个方向把握好分片机制对你来说就不再是黑盒了。
阅读完成 · 觉得有帮助？
