企业官网建设流程全解析

1. 并行度调优：大数据任务提速的第一课

先说个现象。很多搞数据开发的同行，一提到任务慢，第一反应就是加资源：spark.executor.memory从4G调到8G，executor.cores从2加到4，队列从default换到production。资源加上去以后，任务确实快了，但仔细一看，CPU利用率可能连40%都没到，大量executor在空转等shuffle数据，或者在地板上躺平——你花钱买了4核，实际只用上了1.5核。这种浪费，就是因为并行度调优没做到位。

并行度调优，本质上是在回答一个问题：**你的数据切得够不够细，任务分得够不够匀，资源吃得到不到位。**这篇内容我打算把任务并发控制的整个链路拆开讲，从上岸的并行度概念、参数设置逻辑、运行期动态调整，到实际踩过的坑，一次性捋清楚。适合正在搞Spark、Flink、MapReduce这类离线或实时计算任务的同学，也适合刚接手数据平台、经常被“任务跑不动”搞到头大的平台运维。

先抛结论：并行度这个东西，不是越大越好，也不是越小越好，而是“和你的数据量、资源量、任务特性匹配”才是正确。下面我按实际调优的顺序，一步一步说。

2. 并行度是什么，为什么它会成为瓶颈

2.1 从“一条路走到黑”到“多条路分流”

并行度，翻译成人话就是：**你的任务拆成了多少个并发执行的分片。**以Spark为例，一个RDD或者DataFrame会被切分成多个partition，每个partition对应一个Task，由Executor上的一个CPU Core来处理。并行度就是这些partition/Task的数量。

你可以把数据处理想象成搬家：一个100平米的房子，东西打包好了，你叫了一辆货车、一个人搬，车上得来回很多趟，搬一天都够呛。如果你叫了4辆车、4个人同时搬，半天搞定。但如果你叫了400辆车，每辆车就放一个箱子，车堵在楼下排队，人也手忙脚乱——这就是并行度过高的恶果。

核心逻辑很简单：

• 并行度太低 → 资源利用率不足，任务跑得慢；
• 并行度太高 → 调度开销大、小文件多、网络shuffle压力大，反而变慢；
• 并行度匹配 → 数据量与计算能力达到一个平衡，CPU、内存、IO都在干活，既不空闲也不打满。

2.2 并行度的“四层漏斗”

在实际工程里，并行度并不只是一个参数，而是层层递进的四个层级：

这四层是漏斗关系：**输入并行度决定了你能读多快，转换并行度决定了中间处理能否铺满资源，shuffle并行度直接决定数据倾斜和网络传输开销，调度并行度决定了整体能跑多少个Task并发。**调优的时候，这四层必须逐一检查，缺一不可。

打个比方，你有一个600GB的文件存在HDFS上，默认块大小128MB，读取时Spark会生成约4800个partition。如果你的executor有100个core，理论上可以同时处理100个task。但如果某个窄依赖的map操作后紧接着一个groupBy，shuffle.partitions设了200，那么宽依赖后期就只有200个分区了。4800个初始分区汇成200个分区，再把100个core铺上去——前半段吃吞吐，后半段吃并发，中间还夹着一次巨大的shuffle。这个场景就是典型的“漏斗中间收紧”，性能瓶颈往往就在这。

3. 核心参数与匹配逻辑：怎么确定“多少才算合适”

3.1 Spark内存模型和并行度之间的关系

调并行度之前，先得搞清楚一个底层约束：**并行度不能超过CPU可用的并发能力。**每个Task在Executor里就是一个线程，线程数 =spark.executor.cores。假设一个Executor分配了4个core，那么这个Executor最多同时执行4个Task。

Executor内存分三块：Execution Memory（用于shuffle、join、sort）、Storage Memory（缓存RDD）、Reserved Memory（系统开销）。当你把并行度拉高时，每个Task分到的Execution Memory变少，可能会导致spill到磁盘、甚至OOM。这就是为什么“不断加并行度”会在某些任务上报错的原因，哪怕你的CPU还空闲。

举一个真实调参案例（基于常见实践，不是特定公司数据）：

一个6GB的JSON日志清洗任务，集群8个Executor、每Executor 4核，运行时间约17分钟。初始shuffle.partitions是默认200。观察Spark UI发现，Stage 2的Shuffle Read只有2.1GB，却有200个task，每个task平均读的数据才10MB左右——这就是过度分区。后把spark.sql.shuffle.partitions调成80，运行时间压缩到11分钟，约提升35%。原因很简单：200个task中每个task都有一份固定的序列化、反序列化和调度开销，数据本来就少，分区还细，大头全耗在固定开销上了。

反过来也有例子：一个7TB级的大join任务，shuffle.partitions保持默认200，结果每task处理数据约35GB，磁盘spill严重，task跑几小时不结束。调到2000后，运行时间从3小时压到约40分钟。关键是计算task处理数据量，不要拍脑袋。

3.2 数量级估算公式

我总结下来，并行度的良好范围经验公式如下：

期望单Task处理数据量 ≈ 100MB ~ 300MB 并行度 = 总数据量 / 单Task期望处理量

这只是初值。还需要满足另一条约束：

最大有效并行度 = Executor总数 × 每Executor的Core数

如果算出来的“并行度”远大于“最大有效并行度”（比如3000分区 vs 200并发度），那么多出来的partition不会立刻被执行，而是在队列里排队。这不一定是坏事——**如果某个task失败需要重试，排队分片能快速补位，反而不影响整体时长。**但如果分区数超过了最大并行度10倍以上，就说明分区太碎，纯属增加调度负担和元数据压力。

实际操作中，我一般分两步：

1. 先计算数据量和并发度匹配关系：总数据量 /（期望单Task大小）；
2. 再改成按并行度反推：目标并行度 = min(计算出的分区数, 最大有效并行度 × 1.5~3)。

第二步乘以1.5到3，是给“数据倾斜与运行时动态调整”留缓冲。比如某个key数据量大，会造成少数task拖后腿，稍微多一点的分区数可以让其他task更快跑完，等待时间不至于全集中在某个task上。注意：这只是经验值，极端倾斜场景下缓冲效果有限，还需要配合加盐、预聚合等方案，这部分后面细说。

3.3 动态分配：开启开关以后并行度到底还调不调

相当多的集群默认开启了spark.dynamicAllocation.enabled=true。它的逻辑是：初期申请少量Executor，跑一会儿后发现任务排队太多，自动增加Executor；空闲时再回收。

不少同学会困惑：既然能动态伸缩，是不是不用设并行度了？不是。动态分配调节的是Executor数量，而分区数（也就是Task数）不会自动跟着变。也就是说，Executor从10个变成30个，如果shuffle.partitions还是固定在60，那最多只有60个Task可以并行，剩下20个Executor等于白申请。

这里有一个需要特别注意的组合策略：

• 关闭动态分配时：spark.executor.instances必须写死，并行度参数按实例数量 × 核数算；
• 开启动态分配时：spark.dynamicAllocation.minExecutors、maxExecutors要设合理区间，spark.sql.shuffle.partitions必须设成一个偏大的值，让任务数多于最大并发度，才能让动态分配有发挥空间；
• 动态分配下并行度设偏小的危害：明明集群资源充足，任务也一大把，但分区数量不够，导致并发度卡住，跑得比静态分配更慢。

有一个常用的“底线参数组合”，我贴出来供参考：

spark.dynamicAllocation.enabled=true spark.dynamicAllocation.minExecutors=2 spark.dynamicAllocation.maxExecutors=100 spark.dynamicAllocation.initialExecutors=10 spark.sql.shuffle.partitions=500 spark.default.parallelism=200

这个组合的思路是：先给一个较高的shuffle分区上限，让动态分配能够根据实际负载逐步加Executors，同时通过default.parallelism约束默认RDD的初始分区数，避免无脑切分。

3.4 关键参数一览：一表看懂每个参数管什么

这些参数之间不是孤立的，比如spark.executor.memory分配少了，spark.shuffle.memoryFraction再高也没用。所以调整顺序必须是：先确定Executor规格和总量，再定并行度，最后微调shuffle相关缓冲区。

4. 任务并发控制：不只靠Spark内部

4.1 调度层面的并发限流

在大数据架构里，并行度调优的精髓，往往不在单个任务内部，而在任务与任务之间的并发控制。一个数据平台同时只跑一个任务的情况基本不存在。通常是一堆任务挤在同一个队列里，或者是多个团队共享一套集群。

这时候如果不管任务并发，会发生什么？上午10点，上游数据一到位，几十个报表调度任务同时启动。每个任务申请50个Executor，一共需要500个Executor，集群只有200个资源，YARN队列配置的maximum-am-resource-percent也许还限制了AM占用。结果就是：部分任务申请不到资源，在ACCEPTED状态一直挂着；部分任务抢到资源后，因为资源不够，内部Task排长队，整体吞吐反而下降。整个集群看起来CPU曲线很高，但有效产出为零——全在调度和抢资源了。

解决方法是分层限流：

• 队列层面：把不同业务线分配不同的YARN队列，限制每个队列的maximum-capacity和maximum-applications。例如数据仓库团队队列的核心任务最多同时12个应用，临时分析队列最多4个；
• 任务内部：在HQL或Spark提交脚本里，通过spark.executor.instances限制Spark应用自己请求的资源上限；
• 调度层面：使用容量调度器（Capacity Scheduler）或公平调度器（Fair Scheduler），并且配置yarn.scheduler.maximum-allocation-mb和yarn.scheduler.maximum-allocation-vcores，从单Container层面封顶资源请求。

有人说“任务多并发的表象是集群忙，内核是调度器失控”，这个说法我觉得很到位。我以前调试过一个场景：集群800 vcores，某天突然所有任务都变慢。一看YARN管理界面，某个临时脚本的Spark应用申请了400个vcores，而且是--num-executors写死的，也没有动态释放。这个应用本身的并行度设置得很高（1000个partition），但因为只有一个队列，其他任务全排在后面。修复方式很简单：给临时任务划到独立的低优先级队列，并且限制单应用最多4个executor。恢复后其他任务延迟从60分钟降回8分钟。

4.2 实时任务中的并发度控制

批处理说完，说说实时。Flink的并行度控制跟Spark不太一样，它分为：

• 算子并行度：每个算子（Source、Transform、Sink）可以单独设置并行度；
• 全局并行度：env.setParallelism()设置所有算子默认并行度；
• 槽位共享组（Slot Sharing Group）：控制多个算子是否可以复用同一组Slot；
• watermark中的并行度约束：多个并行Source产生watermark需要周期性对齐，并行度越高对齐延迟越大。

常见问题场景是：Kafka某个Topic有64个分区，Flink Source并行度设了16。数据会被4:1地拉取，也就是说每个Flink Source Subtask会连续消费4个Kafka分区的数据。如果其中某个分区消息量大，就会在这个Subtask上形成热点，即使后面Sink并行度设得很高也无济于事。

所以实时任务调整并行度，第一步是Source并行度跟上游Topic分区数保持一致（或者整除关系），第二步是中间处理链的并行度围绕状态大小来定。如果状态很大（例如窗口聚合或去重状态），并行度太高会导致每个算子实例都持有一部分状态，checkpoint持久化时瓶颈变大；并行度太低则本地状态变大，单点压力高。

这里补一个实操经验：**Flink并行度调优，风口不是处理吞吐量，而是状态后端的RocksDB线程数和checkpoint大小。**状态算子的并行度每增加1倍，checkpoint持久化的文件数量就可能增加近1倍（不做增量checkpoint时）。如果你加了并行度但checkpoint时间变长，反而会导致端到端延迟上升。建议并行度调整后，同时观察以下三个指标：

• 吞吐量（records processed per second）
• end-to-end latency（从进入Source到写出Sink的时间）
• checkpoint duration（入库/持久化的耗时）

这三个指标取平衡，不要只盯着吞吐量。

4.3 批流一体视角下的并发控制

很多公司现在都开始做批流一体，也就是同一套SQL逻辑跑批处理和流处理。在这个架构下，并发控制比纯批或纯流都更复杂：

• 离线读全量数据的并行度，和一个实时窗口的并行度，可能差10倍以上；
• 同一个Flink SQL作业切换成批模式运行（通过execution.runtime-mode=BATCH），它的shuffle分区策略又不一样；
• 数据源是文件时，并发度按文件数分配；数据源是消息队列时，并发度按分区数分配。

这中间最容易出的问题是：一套SQL在流模式可以跑，但切到批模式以后，因为table.exec.resource.default-parallelism和流模式的并行度参数不同，导致执行计划里的并行度断层。我建议的做法是：在SQL的Hint里显式指定每个算子的并行度，避免依赖全局默认值。例如：

INSERT INTO dws_result SELECT /*+ OPTIONS('parallelism'='8') */ ... FROM source_table

这种在每个Job上写死并行度，更利于批流切换时保持执行性能特征一致。全局默认值只作为兜底，不要让不同方式切换后行为不可控。

5. 实操记录：一个Spark离线报表任务的完整调优

5.1 任务背景和初始状态

某数据平台有一个离线报表任务，功能是从ODS层读取前一天的全量用户行为日志（文本格式，约620GB），经过清洗、维度表关联、聚合后，写入数仓ADS层。任务每天凌晨跑一次，运行时长经常波动，从2小时到5小时不等。集群资源规模：12个节点的Hadoop集群，每节点16核、64GB内存，YARN总可用资源约192 vcores。

初始提交参数（常见默认，非特定公司）：

spark-submit \ --class com.data.report.UserBehaviorReport \ --master yarn \ --deploy-mode cluster \ --num-executors 20 \ --executor-cores 4 \ --executor-memory 12G \ --driver-memory 4G \ --conf spark.sql.shuffle.partitions=200 \ --conf spark.default.parallelism=60 \ app.jar

前一天的日志文件有约2万个块（128MB块大小）。20个executor × 4核 = 80个并发Task。

5.2 定位瓶颈：从Spark UI入手

跑起来以后，打开Spark UI，我重点看两个Stage：

• Stage 1：读取文本文件、清洗、过滤。这个stage有约8000个Task，每个Task处理约80MB。这个数字是合理的，磁盘读+JSON解析占大头，耗时约9分钟；
• Stage 2：与维度表做广播join（Broadcast Join），然后groupBy聚合。因为有shuffle，每个Task会从上游拉取数据。这个Stage一共只有200个Task（shuffle.partitions=200），但每个Task平均处理约3GB的中间数据。

问题一眼就看出来了：**Stage 1是8000个细粒度Task在跑，Stage 2是200个粗粒度Task在跑。**前一个阶段几乎是以每秒几十个Task并行在刷数据，后一个阶段却是每个Task吃3GB，肯定会慢。

再看Executor的GC时间：20个Executor的GC时间总计占了运行总时间的12%，其中两个Executor的GC高达近30%。说明executor-memory 12G与4核搭配，对于shuffle阶段来说内存偏紧。

5.3 参数调整方案

调优方案，分三部分：

**第一部分：抬高Shuffle分区数。**把spark.sql.shuffle.partitions从200调到460。为什么是460？看数据：中间数据量约620GB（基本是1:1放大），期望单Task 150MB，620GB ÷ 150MB ≈ 4228。用最大并发度约束一下：192 vcores（集群总量）× 2.5 ≈ 480。取400~480之间，所以选460。这样每个Task处理约1.3GB，并发度有460个Task，对于192 vcores来说，排队能接受，最后一批Task不会等太久。

**第二部分：重设executor规格。**原来20个executor × 4核 = 80并发。改成24个executor × 5核，内存仍是12G。核数增加到5，实际上分会稍微超用vcores（24×5=120），但集群192 vcores够用，同时每个Task的内存稍微变少。这里注意，executor-memory 12G如果配5核，每个Task约2.4G内存，其实对shuffle来说还是小。我改成executor-memory 16G，24×16=384GB，但单节点64GB、12节点共768GB物理内存，实际YARN可分配约80%，也就是614GB，384GB在范围内，有富余。单个Executor 16G配5核。

第三部分：调整shuffle读写缓冲。spark.shuffle.file.buffer=128KB，spark.reducer.maxSizeInFlight=96MB。这两个参数属于“经验补强”，在shuffle数据量巨大时，可以减少小碎片写入和读放大。

最终提交参数：

spark-submit \ --class com.data.report.UserBehaviorReport \ --master yarn \ --deploy-mode cluster \ --num-executors 24 \ --executor-cores 5 \ --executor-memory 16G \ --driver-memory 4G \ --conf spark.sql.shuffle.partitions=460 \ --conf spark.default.parallelism=120 \ --conf spark.shuffle.file.buffer=128KB \ --conf spark.reducer.maxSizeInFlight=96MB \ app.jar

5.4 优化结果

• Stage 2的Task数从200变为460，单个Task处理的数据从3GB降为约1.3GB；
• 整体运行时间从2小时58分钟压缩到1小时37分钟，提升约45%；
• GC时间占比从12%降到6%；
• 集群CPU峰值利用率从40%多上升到接近70%（这个值还有提升空间，但已经不死在shuffle上了）。

这个结果不说惊人，但稳定。更关键的是：**调整后任务不再“每天一个样”，运行时长方差大幅减小。**以前5小时和2小时都可能出现，现在稳定在1.5小时上下。这对接上游调度来说帮助很大——下游的依赖任务不再需要预留过长的等待窗口。

5.5 调优过程中踩过的坑

这里多说几个真实的坑：

**坑一：只加Executor不调分区数。**我第一次做类似任务是先加了executor（20个加到40个），shuffle.partitions还是200。结果发现加了资源任务反而慢了。原因就是Task数没变，资源加再多，同一时刻能跑的Task还是最多200个（实际并发80，因为40×2核=80）。而且更多的executor意味着更多的shuffle连接，网络握手开销反而增加。资源加再多，分区数不跟着变，等于白加。

**坑二：在瓶颈不在Shuffle时调Shuffle。**有些任务耗时大头是源端扫描，或者Sink端写入性能差。这时候不管怎么调spark.sql.shuffle.partitions都没用。调优前一定要打开Spark UI看每个Stage的耗时分布，而不是一上来就动shuffle参数。

**坑三：忽略小文件。**如果spark.sql.shuffle.partitions设置得过大，输出Stage的Task也会同样多。比如你算完聚合后写结果表，如果设置的是2000个分区，就会生成2000个小文件。这对下游读取是灾难：HDFS NameNode内存爆掉不说，后续读写速度也断崖式下降。正确做法是最后写数据的时候单独设置coalesce或重新分区，把输出文件数量控制在合理范围。

6. 数据倾斜和热点Task：并发度调优中最隐蔽的敌人

6.1 倾斜带来的假象

并行度调优过程中，很容易被一个假象迷惑：整体Task数量差不多了，为什么运行时间还是没有显著降低？

这时候很可能不是并行度的问题，而是数据倾斜。典型场景：一个join任务，左表大右表小，join key里有几个极热的值（比如某个用户ID每天贡献几千万条行为数据）。shuffle之后，这几个key落在了少数几个Task上，那几个Task要处理的数据可能是其他Task的几十倍。表现就是：大多数Task几秒就跑完了，只剩两三个Task拖在那里，跑十几分钟甚至更久。

如果只看“平均数据量”，你会觉得并行度挺合适；但如果在Spark UI里打开某个Stage看每个Task的Shuffle Read Size，你会发现一个非常刺眼的分布：50%的Task读100MB，5%的Task读8GB。这就是倾斜。

6.2 三种常用倾斜处理方式

1. 加盐（Salting）：把倾斜的key加上随机后缀，分解成多个key，然后在join时先用后缀匹配，再聚合去掉后缀。适合聚合类操作，尤其是groupBy。操作步骤：找出高频key，给这些key加随机前缀生成N个分桶，另一个表对应key加平摊映射，最后聚合。注意：加盐会让结果里有临时膨胀，但能大幅均衡分区负载。

2. 广播小表：如果其中一个表很小（比如几MB），直接广播它，避免shuffle。这能让整条链路大幅提速。实际操作是把小表做成广播变量，或者让Spark自动广播（默认阈值10MB，可调整到更大）。

3. 拆分join并union：如果必须shuffle，可以把数据按key分为“倾斜key”和“正常key”两路，倾斜部分做加盐、加映射后再join，正常部分正常join，最后union。这种方式相对繁琐，但效果最直接。

6.3 用分区策略提前规避

倾斜问题其实可以在并行度设计阶段就预防一部分：如果join key的基数很低，就不要依赖hash partitioning，改用repartition或者range partition去重新散列。Spark的hashPartitioner是按key哈希取模来决定分区的，如果某个key占比高，它天然会落到同一个分区。这时可以尝试salting之后再做hash分区。

另外，如果是日期维度的倾斜（比如月底最后一天数据量暴增），可以考虑在写入源时人为做“二级拆分”。把分区键从“日期”换成“日期+小时”，这样读取时天然有了12或24个子分区，倾斜最多落在一个小时桶里，而不是整个日期桶。

这些手段虽然不在“并行度参数”范畴，但却是并行度能否真正生效的前提——**并行度只是把数据切开，切得好不好、均不均匀，取决于分区键设计。**这就像并行度和分区策略是“刀”和“切法”的关系，刀子快，但切法不对，煮不熟饭。

7. 监控、反馈与自适应调优

7.1 要监控的关键指标

调优不是一次性的，任务数据量每天都在变，集群负载也在变。所以一套可复用的调优链路，必须有监控和反馈。

并行度相关，我建议至少盯住下面几个指标：

7.2 动态调优实践

如果任务量大、类型多，人工逐任务调优根本忙不过来。这时候可以走“基于历史运行数据的动态调参”方案。

基本思路：

1. 在任务提交侧拦截参数，记录每个任务的历史运行数据（输入数据量、输出数据量、耗时、各stage的耗时占比）；
2. 跑一个简单的回归模型，根据输入数据量预测最佳shuffle分区数；
3. 在任务提交时，自动覆盖默认的shuffle.partitions，按预测值注入。

这个方案不需要太复杂的模型，线性拟合或者查表法就有效。我实践过的一个简版方案是这样的：

# 简化版：基于经验公式的动态分区数计算 def compute_shuffle_partitions(total_input_bytes, max_concurrency, target_mb_per_task=150): base_partitions = total_input_bytes / (target_mb_per_task * 1024 * 1024) # 至少要比最大并发度大一些，给长尾留余量 min_partitions = max_concurrency * 3 return int(max(base_partitions, min_partitions))

提交参数里加一行--conf spark.sql.shuffle.partitions=$(python3 compute.py $INPUT_SIZE)，每次数据量变了，分区数也跟着变。比起固定写死200或500，这种方式能在数据量波动大时不至于过度分区或分区不足。

7.3 没有银弹：不同计算引擎的并行度调优差异

不能把Spark的经验直接套到Flink或者MapReduce上。因为：

• Spark是批式并行，shuffle后Task是重新调度的，所以你只管分区数多少，调度器会尽可能均匀地分配到executor上；
• Flink是流式并行，数据在算子之间是流水线传递的，上下游算子的并行度差异会直接导致“背压”（Backpressure），上游处理快、下游处理慢时，数据会在上游堆积。没有shuffle那一下重新切分，所以必须保证Sink算子并发度不低于Source；
• MapReduce的并行度主要靠输入切片决定，且map和reduce之间没有动态分区调整，reduce的并行度只能在提交时定死。

不同引擎的并行度调优，都有各自的一条主线。但这篇的核心还是Spark为主，毕竟大数据架构里Spark用的最广，遇到的坑也最深。

8. 调优工具与辅助手段

8.1 Spark UI怎么用才高效

调优的第一现场就是Spark UI。但很多人只是看一下“运行时间”和“成功/失败数”，这远远不够。有几个地方我建议每次调优都必看：

1. Stage列表页：看每个Stage的耗时，先找耗时最大的那个，再动手。盲目调参数之前，至少要知道瓶颈在哪一个stage；
2. 单个Stage的Task详情：按“Shuffle Write Size”和“Shuffle Read Size”排序，快速判断是否倾斜。倾斜通常一眼就能看出来；
3. Event Timeline：看每个Executor的开始/结束时间和Task的排队等待时间，如果大量时间花在等待（黄色块），说明资源不够或分区数少于并发度；
4. Executors页面的GC时间：GC时间占比高意味着内存不足。要么加大spark.executor.memory，要么减少spark.executor.cores。

8.2 几个好用的调优辅助脚本

下面是一个小脚本的思路，用来自动分析Spark日志中的shuffle数据：

#!/bin/bash # 从Spark History Server日志中提取每个stage的shuffle读写信息 # 用法：analyze_spark_stages.sh <app_id> APP_ID=$1 for stage_id in $(curl -s "http://<history-server>/api/v1/applications/${APP_ID}/stages" | jq -r '.[].stageId'); do echo "Stage $stage_id:" curl -s "http://<history-server>/api/v1/applications/${APP_ID}/stages/${stage_id}" | \ jq '{shuffleRead: .[0].shuffleReadBytes, shuffleWrite: .[0].shuffleWriteBytes, numTasks: .[0].numTasks}' done

这个脚本的价值在于，批量分析历史应用时不用一个个页面点开，直接拿到各stage的shuffle数据量和task数量，再配合计算工具判断分区是否合理。

8.3 调优和架构设计的关系

并行度调优做到一定阶段，你会发现纯粹调参数已经不能带来质变了。如果任务已经在合理并行度下运行，却发现仍然很慢，那就要回头想想架构本身：

• 数据源是不是需要分区？Kafka的Topic分区数够不够（通常建议分区数≥消费者并行度）；
• 存储文件是不是小文件过多？如果源表有几十万个几KB的小文件，并行度再均匀也会被metadata开销拖死；
• Shuffle模型能不能避免？如果join的key分布本身就倾斜，能不能在数仓模型设计时通过预处理消除倾斜？

在这些情况下，**并行度调优只是缓解症状，架构设计才是治本。**比如，一个大表join另一个大表的场景，如果先把大表按业务维度和先聚合掉一部分数据，再做join，整体shuffle量可能下降90%，这比任何并行度参数都更有效。

9. 常见问题与排查快查表

把实际工作中最常遇见的并行度问题整理成一个速查表，方便遇到问题的时候直接对号入座。

9.1 问题速查表

9.2 定位问题的标准动作

遇到一个跑得慢的任务，我一向按下面的动作链排查：

1. 打开Spark UI，看Stage列表，找到耗时最长的那个Stage；
2. 点进该Stage，看Task数量分布、平均处理数据量、最长Task和最短Task的比例；
3. 看Executor页的GC和内存曲线；
4. 看YARN或资源管理界面，确认当前集群整体负载和排队情况；
5. 结合以上信息，确定瓶颈是资源不够、分区不合适、倾斜，还是代码本身的问题。

这个动作链非常机械，但每次都是它帮我快速定位问题。调优不是玄学，而是一个个指标排除后剩下的那个原因。

10. 写在最后：并行度调优真正教会我的事

并行度调优表面上是一堆参数的配置，本质上是理解你的数据形状和计算资源之间的关系。很多人以为调优是把参数往大了调，实际上反而是要找到数据量和执行单元之间的平衡点。

我在实际工作中最大的体会是：**并行度调优没有“万能参数”，只有“对症下药”。**同样一份数据，在A集群上最优的shuffle.partitions拿到B集群上可能一塌糊涂；同样一个任务，今天跑得好，明天因为上游数据量变化，又得重新调。

所以我最后有几个小建议：

1. 每次调优前，先记录当时的运行时间、资源部署、数据量，调完以后对比，别凭感觉说“变快了”还是“变慢了”；
2. 把自己调优过的任务参数沉淀成模板，在面对同类任务时先套模板，再根据实际运行数据微调；
3. 跟踪一段时间的运行数据后，把参数计算逻辑自动化，让系统自己根据输入数据量动态生成分区数，这条路比每次手动调优更省心。

还有一个很容易被忽略的小技巧：**调完参数以后，跑一次，截个图，隔两周再跑一次，对比两次运行时长。**如果两周后的运行数据和第一次差不多，说明你的调优结果是稳定可复现的；如果不稳定，那集群资源竞争或者上游数据波动才是真正的问题。千万别只跑一次就拍胸脯说搞定了。

数据量永远在变，资源永远在争，架构永远在演进。掌握并行度调优的方法论，比记下一堆参数值更有价值。