大数据架构里的并行度调优说到底就是一件事在有限的计算资源下把任务拆分得刚刚好让每一份资源都被高效利用而不是互相等待或抢来抢去。很多刚接触分布式计算的人一上来就被“并行度”“分区数”“并发度”“资源规格”这些词绕晕以为把它们都调大就是优化。结果呢资源翻倍任务不但没变快反而更慢了——调度开销变大网络传输变多小任务排队时间甚至比计算时间还长。我见过不少因为这个踩坑的案例集群里CPU飙得老高但Spark作业的进度条就是纹丝不动。本文就围绕大数据架构中的任务并行与并发控制把并行度背后的原理、常见的调优思路、实操中会遇到的问题一次性说清楚。适合正在做离线数仓、实时计算或者刚接触分布式计算、准备好好调一调自己作业的人。1. 并行度到底调的是什么1.1 并行度、并发度、分区数的区别这三个概念容易混淆但本质完全不同。并行度Parallelism指的是在同一时刻能够同时执行的计算任务数量。比如一个Spark作业你能同时跑多少个task这个数值就等于执行器executor的资源槽位总和。并行度决定了计算的“宽度”。并发度Concurrency在大数据语境下更侧重于指同时处于运行状态的作业job或任务task数量。它关注的是有多少任务在同时推进而实际是否能同时执行受限于并行度。分区数Partitions是数据在计算框架里的物理划分。RDD的分区数直接决定了task的数量一个分区对应一个task。所以很多时候我们所说的“提高并行度”实际上是在调整分区数。从机制上看它们的关系可以这样理解你在任务层面设置了高并发但如果机器的CPU核数和executor数量不够任务就会排队你增加了分区数但如果运行资源没跟上同样会在调度器那里卡住。调优的本质是在这几个变量之间找到那个平衡点。1.2 一个任务的完整执行链路要理解并行度调优得先看清楚一个任务从提交到运行经历了什么。以Spark为例作业提交后Driver会将作业拆分为多个job每个job再按shuffle或stage边界拆分为stage每个stage内部进一步拆分为task。这些task由Driver分配到各个节点的executor上执行。关键点是task并不是提交了就能马上运行的。executor里有可用的计算槽位core每个槽位同一时刻只能运行一个task。如果executor数量少或者槽位被占用剩下的task只能排队等调度。所以并行度的本质是你的数据切分粒度 × 你的计算资源可用性 实际运行效率。实际调优时我习惯先把两个问题分开看数据层面分区数够不够有没有数据倾斜资源层面executor数量、核数、内存是否匹配分区数。资源够了再调分区数分区数合理了再看倾斜很多看似“并行度不够”的问题其实出在资源与分区的匹配度上。2. 核心参数配置与原理2.1 Spark作业并行度的参数层级Spark的并行度配置涉及多层而且每一步都有自己的独立参数。很多人调的并行度不生效就是因为参数层级用错了。第一层是配置全局默认并行度spark.default.parallelism。如果不显式设置对于RDD操作它就是所有executor上总核数的2倍旧版本或者根据具体操作自动推断新版本。第二层是应用级别。每个RDD本身的分区数由数据源和操作决定。从HDFS读文件时分区数取决于文件块数与InputFormat的切片逻辑调用repartition(n)、coalesce(n)、groupBy、join这类宽依赖算子时分区数会发生变化不同的算子对应的shuffle分区数参数也不同最常见的是spark.sql.shuffle.partitions默认200。第三层是执行阶段。每个stage的task数量由最终RDD分区数决定一个stage的task上限受限于executor的资源槽位总数。并行度设置得再高如果资源总数不足task也会排队等待。所以调并行度的起点不是去改任一参数而是先想清楚我的数据量有多大executor能提供的总并发执行能力是多少2.2 不同计算引擎的参数对比大数据架构里不只有SparkFlink、MapReduce、Hive都会有类似的并行度设置其实质同样是控制资源与任务的匹配。它们的核心参数如下。引擎核心参数作用范围备注Spark SQLspark.sql.shuffle.partitions每个shuffle阶段的默认分区数默认200大作业通常不够Spark RDDspark.default.parallelism无shuffle操作的默认并行度小文件或纯转换场景适用Spark 动态资源spark.dynamicAllocation.enabled根据负载动态增减executor开启后需配套设置单executor核数Flinkparallelism.default算子的全局默认并行度可在算子链关键节点精细设置并控制Flinkexecution.buffer-timeout并发任务间的数据传输超时微批时增大可降低交互瓶颈Hive/MapReducemapreduce.job.reduces控制reduce阶段并发通常设为核心数的1~2倍Hivehive.exec.parallel并行执行stage需谨慎会挤占资源2.3 动态分配与静态分配的选择这里有一个很实际的分歧是预先分配好固定资源还是让框架按负载动态调整静态分配的优点是稳定、可预测。适合周期性ETL任务数据量波动不大时固定资源加上合理的并行度设置每个批次的耗时都很平稳方便做SLA保障。动态分配的优点则是资源利用率高。适合数据量波动明显的场景数据高峰时自动增加executor低谷时减少executor避免占用闲置资源。我的做法是分场景数仓核心链路、需要稳定产出时间的任务用静态分配探索性分析、临时脚本、报表类任务开启动态分配。动态分配里有几个细节值得注意设置最小和最大executor数量时通常我会把最大值设为所需峰值的120%shuffle服务的相关参数需要额外配置否则在动态调整executor时可能因中间文件无法访问而报错队列资源的限制必须提前确认否则动态分配无法扩展。3. 实操调优从环境到参数3.1 确认集群资源与任务上限调优前先摸清家底否则调来调去只是隔靴搔痒。我做过一个小实验来理解“核数”与“并行度上限”之间的关系一台8核的机器上启动两个executor每个executor是2核意味着每个executor可以同时运行2个task总计并发能力是4个task。然而一旦task数量到达5个以上就有task在排队等待空闲core了。这里要注意几个常见的资源参数匹配陷阱--num-executors总的executor数量。如果设得过多超过了队列最大资源配额任务根本提交不上--executor-cores每个executor的核数。这个值设置太高会导致碎片化资源难以分配到整节点--executor-memory过大时容易触发OOM过小时任务处理能力弱集群通常还有单用户的最大资源限制超过后同样会被拒绝或阻塞。我的建议是先用集群监控页面查看当前队列的可用资源、已被占用的资源。比如一个队列总共有100个vcore已被占用60个那么你最多只能用40个。在这个基础上并行度要与此对齐。3.2 估算合理的任务并发上限一个可以直接参考的计算思路是可用执行槽位核数 单任务可申请的总核数上限在Spark中排队的task数量往往会积压在executor的调度队列里。通过Spark UI的“Active Stages”页面可以直观看到每个stage的task数量和已完成task数量。如果你发现一个stage里90%的task都完成了还有10%的task一直没跑完大概率不是并行度问题而是数据倾斜问题——少数key对应的数据量远超其他key导致个别task负载过重。如果想要单纯估算并行度的“合理上限”可以先这样算您的核心executor核数总和比如10个executor每个3核那么并行执行上限就是30个task对于数据量大、无shuffle的纯扫描任务我习惯让task数 资源的23倍这样做是为了让CPU在task切换时能更好的隐藏IO等待对于有shuffle的任务shuffle分区数则是多少就会决定下一个stage的task数量因此结合下游的处理逻辑一起推敲上下游才能得到均衡。3.3 一个示例从240个分区到120个分区的调整以前处理过一个大表join场景数据量约2TBSpark SQL默认shuffle分区数200运行时长一直稳定在55分钟左右。任务配置是executor 20个每个5核总并行槽位100个。问题很明显200个task却只有100个槽位等于一半的task在排队而且由于每个task处理的数据量较大GC时间不短。调整思路很简单把spark.sql.shuffle.partitions从200降到120让task数和槽位数匹配同时每个task处理的数据量略有增加。结果呢GC时间从13分钟降到了7分钟总时长从55分钟降到约32分钟。原因在于分区数从200降到120后每个task处理的数据量更大了但并没有超过单task的合理处理范围反而减少了task调度与序列化的开销。这说明并行度调参时并非越大越好而是要让每个task的工作量与它的运行资源相匹配。3.4 SQL层面控制并行度部分参数主要影响执行计划。拿Spark SQL来说除了调spark.sql.shuffle.partitionsSQL本身的写法也会改变并行度表现。子查询与CTE如果一段SQL里一个CTE被引用多次Spark会重复计算。在某些情况下可以增加一个中间表并写缓存cache可以让后续读取并行化从而减少重复计算带来的等待小表与大表join如果不小心关掉了自动广播会发生完整shuffle。手动加/* BROADCAST */提示能避免放大下游shuffle的task数量group by 与 distinct都会触发shuffle如果group by的key区分度不高就会遇到单个task处理量过高也就是倾斜。可以在SQL里给key加随机前缀做两阶段聚合一定程度上缓解倾斜。SQL改写带来的效果非常直接也是并行度调优中“不需要改配置就能提升性能”的高性价比路径。4. 资源竞争与任务并发冲突4.1 多任务同时跑时的资源分配问题在真实的生产集群里不太可能只有你一个人的任务在跑。同一个队列中的多个作业并行执行时相互之间的资源竞争才是常态。这也正是“任务并发控制”中容易被忽视但又非常影响稳定性的部分。两个任务同时跑每个都申请最大资源结果导致谁都无法拿到足够的资源双双阻塞这种情况在共享集群里非常常见。针对这类场景比较有用的调控方式有以下几种配置任务优先级给关键链路任务设置较高的优先级让调度器在资源紧张时优先保障限制单个作业的资源上限而不是放任它去无限申请相当于给并发控制设置了边界用队列隔离不同团队、不同业务设计不同的资源队列避免相互挤兑开启公平调度这类调度模式下多个作业可均匀获得资源而不是第一个作业占满全部资源后第二个再等待。4.2 控制并发度的常用手段设置并行度参数是控制单个作业内并发度的方式控制整个集群中的任务并发程度则要考虑更多限制同一时间提交的作业数量。很多平台支持全局调度限制同一队列下同时运行的作业数上限设得太大会导致资源争抢设得太小则导致资源闲置。控制每个作业的并行度上限。同一队列总资源固定每个作业的并发度之和不可能无限增长提前规划好各作业的并行度上限是避免资源超卖的唯一方式。错峰执行主要作业。将不同团队的跑批时间错开避免高峰期全部挤在一起。这是最简单并且见效快不过也是执行成本高的做法。4.3 动态调整的时机与策略生产环境中的资源常常随时间变化比如早晨的报表任务少午后的数据同步任务多夜间则集中跑批。一成不变的并行度配置很难适应这种波动因此很多团队会选择动态资源机制。我在实际使用中有几个判断原则平均CPU使用率长时间低于20%时可以适当调小资源节约成本当任务的平均排队时间超过单任务本身的执行时间时说明并行度已经严重超过资源承载量需要主动降并行度或扩容动态调整要留有缓冲避免频繁上下波动造成不必要的资源浪费。5. 常见问题与排查技巧5.1 并行度明明调高了任务反而更慢这个现象第一次遇到时可能需要花点时间才能想明白应该是更快的呀从原理上解释并行度提高意味着task数量增多每个task处理的数据量变小但task调度、序列化、shuffle读数据的开销都是固定成本。一个100MB的数据块切成2块和切成10块后者的总耗时未必缩短因为会有更多的网络与磁盘IO来传递那些小的分片。所以遇到“并行度调高但更慢”的情况先做几件事在Spark UI里看stage的“Shuffle Read Size/Records”确认单个task的shuffle数据量是不是太小看“Scheduler Delay”如果明显增大说明是调度瓶颈看“GC Time”如果GC时间占比增长说明内存压力变大。这三种情况都指向了并行度调过头——单个task的数据量过小固定开销占比过高。解决方法也很直接适度降低并行度或者增加单task的处理数据量。还有一个背锅重灾区是小文件问题。如果上游每天的产出物是几千个小文件你读入时分区数天然就很大再强行调低并行度也执行不下去因为HDFS切片逻辑会生成很多空分区白白浪费计算资源。这种问题要从上游治理用repartition或coalesce及时合并小文件。5.2 数据倾斜导致的“长尾task”并行度调优遇到的另一个大坑是数据倾斜即个别task处理的数据量远大于其他task。表现很典型一个stage里绝大部分task几十秒就完成了只有个别task跑了十几分钟甚至更久。排查方法观察Spark UI中task的数据分布找到运行时间最长的那几个task确认这几个task的输入数据量是否为其他task的数倍以上如果确认是倾斜再对key的分布情况做具体的分析最常见的是key为null或者某个热点key的值。常用的解决手段加盐salting给倾斜的key加随机前缀将大key拆成多个子key两阶段聚合先局部聚合再全局聚合适用于group by类型的倾斜广播小表如果join的一端很小直接触发broadcast join绕过shuffle阶段的倾斜过滤空key如果是业务上的无效数据直接清理掉。5.3 任务排队与资源不足的排查路径当你提交任务后发现作业一直处于“Accepted”或“Pending”状态而不是正常执行大概率是资源问题。排查路径查看队列的资源使用情况确认是否有可用资源查看你的作业申请的资源量是否超过了队列配额检查是否同时跑的作业太多把队列占满了如果上面三步都正常再看调度器配置确认优先级或超时相关设定。资源不足时我一般会做一件事从YARN的“Running”和“Pending”页面对比看是vcore不足还是内存不足。很多时候队列内存耗尽但vcore还有很多空闲说明有人内存申请过大了。这类问题用资源隔离或限制单个任务内存上限来控制可以在一定程度上规避。5.4 一个印象深刻的参数调优案例有一次处理一个明细数据统计任务输入数据约500GBHive里跑全量聚合执行时间一度超过2小时一直查不出原因。查看执行计划后发现有几处join用到了大表join大表默认触发全shuffle产生了几千个task。但executor资源只有30个vcore导致一个stage里同时运行的任务也就只有30个剩余tasks全部排队。我没有急着调整并行度而是先分析了两张大表的关联字段发现可以先将其中一张大表的数据提前聚合过滤让数据量减少到原来的五分之一再做join。同时针对性设置并行度到90使得每个task的负载与可用核心数在合理比例上。改动后的作业运行时间下降到了约40分钟。这个案例给我们的经验是并行度调优首先从减少数据量开始再考虑增加资源或task数量。6. 经验心得与避坑建议6.1 调优的层次化思维并行度调优不是单纯调一个参数。我在实操中逐渐形成了一套顺序基本能覆盖大多数场景先看数据规模每天增量多少总数据量多少是否分布均匀再看资源情况集群空闲情况、分配配额、同一时间并行的其他任务情况然后调并行度根据资源和数据规模综合设定task数与并发度最后验证和观测运行过程中的稳定性、有无倾斜和长尾再回去微调。把这个顺序固定下来比拿到参数就“调大试一下”要靠谱得多也更好排查问题。6.2 逆经验——过度调优的伤害调整并行度时“过度调优”是一种很隐蔽但客观存在的问题。常见做法是把并行度设成很大觉得这样总能利用更多资源但在某些类型任务中反而会适得其反。尤其是迭代计算、频繁切换状态的作业并行度过大时通信和状态管理的开销会盖过计算收益。在这种场景我反而会把并行度下调让计算更集中减少数据在节点之间的传输往往效果更稳定。6.3 监控指标要盯哪些最后分享几个我在调优时重点盯的监控指标新上手的朋友建议先熟悉它们输入数据量确认每个task处理的数据量是否在合理范围一般在100MB~1GB之间比较合适Shuffle Read Sizeshuffle读的数据量过大说明分区数或数据分布存在不合理GC时间GC占执行时间的比例超过10%就要考虑增加内存或者降低并行度调度延迟如果task运行时长期处于等待调度状态并行度超过资源承受能力了执行时间分布看是不是有长尾长尾则是倾斜不是并行度调整能直接解决的。做并行度调优别老想着“越大越好”或“越多越好”核心逻辑是匹配——数据量、资源槽位、task数量、单task负载这几个维度匹配上了任务自然又快又稳。我自己的习惯是把每次调优的参数、效果、原因记录下来前几次可能要靠经验和大量试验但积累多了之后基本看一眼任务规模就能大致判断出合理的并行度区间了。