数据怎样移动，决定了执行计划的边界
同样是处理一万行数据，过滤可以在各分区内完成，按键汇总却需要重新组织数据。增加任务数量之前，先看哪些数据必须相遇。
分布式计算把计算分散到数据所在处，再将有共同处理条件的数据放到一起。MapReduce 的 map、shuffle、reduce，Spark 的依赖图与 stage，都在回答数据由谁处理、什么时候移动、失败后重做什么。
本文用 Spark 4.2.0 的 local[2] 模式验证计划、分区行数、显式缓存和结果等价性。容器限制为 2 CPU、3GiB 内存；没有跨机器网络、分布式存储或吞吐测试。本文的局部结果不能推导集群扩展比例。
一、先找必须重新分配数据的地方
对一个分区里的每行做转换，通常不必先读取其他分区。按键聚合、某些连接和重新分区，则需要把相同键的数据送往相应的处理分区。
窄依赖（Narrow Dependency）描述父分区与子分区之间较局部的依赖；宽依赖需要跨分区的数据交换。Spark 沿依赖组织 stage，shuffle 常成为其边界。DataFrame 还经过优化器和物理规划器，不能把每一行源码、每个算子或每个 Exchange 简单等同于固定数量的最终 stage；广播、复用和自适应执行都要看实际计划。
图中第二段不只改变计算位置，还增加分区组织、序列化、传输、缓冲或落盘和失败重做的工作。本机运行也会发生 shuffle，不能把它只理解为网络传输。
MapReduce 的 map 按输入分片执行，shuffle 将记录按键分组并排序后交给 reducer；可用的局部合并能够减少传输，但不能改变计算含义。Spark 可用更一般的 DAG 表达多个阶段，并按需保留中间结果，既有内存操作，也有磁盘 shuffle。框架名本身不能给出固定性能倍数。
二、从实际计划里看局部处理和 Exchange
实验创建 10,000 行、4 个输入分区；前 8,000 行的键为 0，其余行分散在 100 个键中。以下片段来自可运行脚本，省略 SparkSession 和造数：
narrow = base.filter("id % 2 = 0").select("id", "key")
grouped = base.groupBy("key").count()
narrow.explain("formatted")
grouped.explain("formatted")实验显式关闭 SQL 自适应执行，设置 4 个 shuffle 分区，以便比较固定计划；这不是调优建议。输出中的关键关系为：
过滤与投影：Project → Filter → Range
分组计数：HashAggregate → Exchange → HashAggregate → Project → Range这是把原始树压成单行的阅读摘录，从右向左表示数据流。归档保存完整 explain 输出。计数计划先做局部 partial_count，再按键交换局部结果，最后合并；并不一定将热点键的全部原始行都传给一个 reducer。
脚本同时归档事件日志与 stage 信息，并从真实 TaskEnd 指标累计得到正数的 shuffle 写字节。该数覆盖脚本中的多个 action 和不同场景，不能当成上面这一个查询的传输成本。stage 数量和耗时也不用于性能断言。需要诊断某个线上作业时，应按 execution/job/stage 对齐事件和 SQL 指标，不能混加整个应用的所有任务。
三、热点键需要观察交换之后的分布
为避免局部聚合掩盖输入倾斜，实验先直接 repartition(4, 'key')，再统计每个结果分区的行数；对照方案增加由 id % 16 生成的 salt，按 key, salt 分区：
| 场景 | 四个分区的行数，按大小排序 |
|---|---|
| 只按 key | 480、500、560、8,460 |
| 按 key 与 salt | 1,950、1,955、2,005、4,090 |
固定数据、版本和配置下，热点分区从 8,460 行降为 4,090 行。哈希仍会碰撞，16 个 salt 也不会平均映射到 4 个物理分区；这里没有宣称「均分」，也没有测得执行更快。断言只要求最大分区下降。
拆散一个键以后，还要恢复原问题的语义。实验对 (key, salt) 计数，再按 key 求和，逐键比较未加盐结果，101 个键的计数完全一致，总数仍为 10,000。加盐带来额外聚合与数据交换，是否值得要测实际瓶颈。
计数、求和这类可组合的聚合适合分段合并；平均值需要携带总和与数量，不能把各组平均值再次平均。精确去重、顺序敏感计算和连接，还需要专门处理去重范围、顺序或另一侧复制成本。先证明语义等价，再谈并行度。Spark SQL 调优文档还区分了自适应合并分区、倾斜连接等机制，不能把它们当成所有热点聚合的自动解决方案。
四、缓存需要声明、物化和退出条件
构造 DataFrame 不会自动缓存它。实验检查新建的 base.is_cached 为 false；调用 cache() 并以 count() 触发计算后，后续过滤计划中出现 InMemoryTableScan；unpersist() 后缓存标记恢复为 false。
base.cache()
base.count()
# 使用同一份中间结果的后续计算
base.unpersist()缓存有存储级别、容量和驱逐成本，分区丢失后也可能需要重算。它适合重复使用且计算成本足够高的中间结果；只有一次使用、结果过大或输入持续变化时，收益并不成立。缓存可重算的数据也不能当成保存外部副作用的事务记录。持久化级别与 shuffle 行为分别见 RDD 指南及 shuffle 说明。
从计划排查时，先定位 Exchange 的原因和输入，再按 stage 查看分区行数、字节、spill 与长尾任务。最后才考虑改键、局部预聚合、加盐或调整分区数。任务依赖与失败传播的通用模型见依赖图执行；有状态流任务如何恢复及结果何时可见，继续读状态恢复与在线结果。
配套实验
- codesphere-labs/distributed/execution-plan-shuffle：Spark 4.2.0 local[2]：完整物理计划、事件日志、实际分区计数、加盐后聚合等价性与显式缓存；无吞吐结论（验证记录）
参考资料