算上这一篇“开源组件系列”已经写到了第十二篇。前面聊过不少存储和调度相关的组件这篇来把计算引擎这块的核心拼图补上Spark。说实话现在出去面试或者接大数据相关的项目Spark基本上是跳不过去的一环。很多人对Spark的理解停留在“比MapReduce快很多的内存计算框架”但真要问到它为什么快、DAG模型到底解决了什么问题、一个作业从提交到出结果在集群里经历了什么能讲清楚的人其实不多。这篇不打算照着官方文档翻译一遍。我想从“为什么要用DAG来组织计算”这个底层逻辑开始把Spark的作业调度、运行机制、三套核心API的区别、部署形态以及我在实际项目中踩过的调优和排坑经验串起来聊。适合两类人看一类是刚接触Spark、想系统建立认知框架的开发者另一类是已经写过不少Spark代码、但总觉得调度和优化部分隔着一层窗户纸的同学。1. Spark靠什么接过了大数据计算的大旗1.1 在Spark出现之前批处理是怎么跑的要理解Spark的价值得先回到它之前那个时代。早期的大数据批处理主流方案是基于MapReduce模型的。MapReduce把计算过程强行拆成两个阶段Map阶段和Reduce阶段中间隔着一次落盘和网络传输。Map产出的中间结果必须写到本地磁盘等Reduce阶段启动后再去拉取。这种设计最大的问题在于一个复杂的分析任务通常需要串联很多个MapReduce作业前一个作业的输出是后一个作业的输入每一轮都要经过“写磁盘、网络传输、再读磁盘”的过程。单纯从这个流程就能看出来磁盘IO和序列化开销会占据整个作业执行时间相当高的比例。如果是一个需要十几轮迭代的算法比如一些图计算或者机器学习类的任务每一轮都把中间结果刷到磁盘性能会非常难看。这本质上不是MapReduce模型本身算错了而是它对“中间数据”的处理方式太浪费。1.2 DAG把“一步一落盘”变成了“内存里接力”Spark对这件事最核心的改动有两点引入有向无环图DAG来组织计算流程以及把中间结果尽量驻留在内存中。先说DAG。一个数据处理任务从数据源到最终结果中间涉及的各种转换操作比如过滤、映射、聚合、关联这些操作之间的依赖关系天然就能表达成一张图。Spark把这整张DAG交给调度引擎调度引擎对图进行解析找出哪些计算可以放在同一个阶段Stage里串行执行哪些阶段之间存在shuffle依赖必须分开然后以最优的方式把任务分发到集群里的各个节点上去执行。这跟MapReduce那种“机械地拆成固定两段”完全不同DAG的灵活性在于它允许一个阶段的结果直接被下一个阶段消费中间省去了大量落盘。配合内存缓存机制那些需要反复迭代的计算就能在内存里连续接力完成。这也是Spark在迭代式计算和交互式查询场景下比老方案快出几个数量级的根本原因。1.3 内存计算不是越猛越好谈到这里需要纠正一个常见的误解Spark并不是完全不需要磁盘。当内存不足以存放全部中间数据时Spark仍然会把溢出部分写到磁盘也就是常见的spark.shuffle.spill机制。内存计算的意义在于“减少不必要的落盘”而不是“完全消灭落盘”。理解了这点后续做内存参数调优时就不容易走极端。另外有一点很多人在初学阶段容易忽略Spark虽然比老方案快但它不是一个“内存数据库”那样的系统。它的核心依然是一个分布式计算框架只是通过DAG调度和内存复用来大幅缩短了作业的执行时间。搞清楚这层定位对后面我们讨论资源参数配置会很有帮助。2. DAG计算模型的底层拆解2.1 RDDDAG的节点与边Spark的DAG不是凭空产生的它的构建基础是RDD弹性分布式数据集。我们可以把RDD理解为一个“带分区信息的不可变数据集”它记录了数据分布在哪些节点上以及对这个数据集已经施加过哪些转换操作。在Spark里用户的程序基本上就是在不断地创建新的RDD并对其进行转换比如从一个文本文件创建RDD然后map、filter、join每一步操作都会生成一个记录了“我这个数据集是从哪个父数据集、经过什么计算得到的”的RDD。这一系列RDD之间的依赖关系串起来就是一张血缘关系图Lineage Graph。Spark的DAG调度器拿到的正是这张血缘图。这里值得强调的是“不可变”这个特性。RDD一旦创建就不会被修改任何转换操作都生成新的RDD而不是改动原有数据。这个设计带来一个巨大的红利如果某个分区的计算失败或者数据丢失Spark可以沿着血缘关系回溯只重新计算受影响的那个分区而不是把整个作业从头跑一遍。这就是容错机制的基础。2.2 窄依赖与宽依赖划分Stage的标尺光有图还不够调度器要执行这张图必须搞清楚哪些计算能“就地串联”哪些计算必须“停下来等待数据重排”。这就需要区分两类依赖窄依赖是指父RDD的每个分区最多被子RDD的一个分区使用。典型的操作就是map、filter、union。窄依赖场景下父分区的计算和子分区的计算可以在同一个任务里连续完成不需要跨节点传数据效率极高。宽依赖是指父RDD的每个分区可能被子RDD的多个分区使用典型的就是groupByKey、reduceByKey、join这一类会触发shuffle的操作。宽依赖存在时必须等父阶段的全部任务执行完然后按照key重新分区和传输数据才能开始下一阶段的计算。DAG调度器划分Stage的规则很简单粗暴从后往前回溯遇到宽依赖就把依赖链切断切出来的每一段就是一个Stage。换句话说Stage的边界就是shuffle发生的位置。2.3 一张图看懂执行顺序下面用一个小场景把这几个概念串起来。假设有这样的一个计算逻辑读取两份数据集一份做过滤并聚合另一份做映射然后两者做关联最后排序输出。在实际执行时Spark先生成DAG然后DAG调度器发现关联操作join是一个宽依赖因此它被设为Stage边界关联之前的所有操作会被压缩到前一个Stage里关联之后直到排序输出则属于后一个Stage。同一个Stage内部的任务之间不需要数据重排可以尽可能地利用内存和CPU流水线式执行。这个流程实际对应的执行计划是DAG操作是否触发shuffleStage归属读取源数据否Stage 0过滤、映射、局部聚合否Stage 0关联操作是Stage 0与Stage 1的边界排序与结果输出是Stage 12.4 DAG相比两阶段模型的优势到这里可以做个明确的对比了。MapReduce模型中每个业务逻辑都被强行套进Map和Reduce两个阶段里复杂任务需要手动拆成多个作业串行执行层层落盘。DAG模型中计算图在调度层是全局可见的调度器能自动识别哪些计算可以合并执行哪些必须分开然后把尽可能多的计算放进同一个Stage内完成。这种“全局视角”带来的直接收益是对于多步骤任务作业数量从原来的“每个MapReduce作业都要完整走一遍提交、调度、执行、落盘的流程”变成了“一次提交内部自动划分阶段”。作业粒度和调度开销都大幅下降尤其适合那些需要几十步转换的复杂ETL逻辑。3. 一次Spark作业在集群里的完整旅程3.1 Driver、Executor和集群管理器的三角关系理解DAG调度模型之后有必要看看一个作业提交后到底经过了哪些环节。Spark采用经典的主从架构Driver负责解析代码、构建DAG、划分Stage、调度TaskExecutor负责真正执行Task并在内存里缓存数据。集群管理器可以是自带的Standalone模式也可以对接YARN或者Kubernetes负责资源的分配和回收。有个概念容易混淆Driver和集群管理器各自管什么。Driver管的是“计算任务的编排”它知道要执行哪些Task、每个Task的数据在哪集群管理器管的是“计算资源的分配”它只知道有多少个Executor可以启动每个Executor给多少CPU和内存至于Task在Executor内部怎么跑它不关心。3.2 作业、阶段、任务三层调度结构Spark的调度是典型的三层模型。一个完整的作业Job是用户提交的一个动作比如一次count操作或者一个SQL查询一个作业会被拆成多个阶段Stage阶段之间以宽依赖为界每个阶段对应多个任务TaskTask的数量取决于分区数每个分区由一个Task负责处理。这三层关系值得多说明一下。Task是实际在Executor上运行的单元一个Executor默认情况下可以并发运行多个Task具体取决于分配的CPU核心数。比如分配了4个核心那这个Executor上就能同时跑4个Task。分区数和并行度直接挂钩这个数设置太小会导致资源利用率不足设置太大又会增加调度开销。3.3 数据本地性让计算主动靠近数据Spark调度还有一个容易被忽略但影响很大的策略数据本地性。它的基本原则是“能不传数据就不传数据尽量让Task到数据所在的节点上去执行”。因为网络传输的代价通常比在节点本地读数据要高得多尤其对于TB级别的大数据集数据的移动会造成巨大的IO和网络开销。本地性由高到低分为几个等级进程本地性PROCESS_LOCAL数据就在同一个Executor里、节点本地性NODE_LOCAL数据在同一个节点的其他Executor里、机架本地性RACK_LOCAL数据在同一个机架的其他节点上等。调度器会先尽量满足高等级本地性如果等待超时还拿不到对应位置的资源会降级执行从其他节点拉取数据。3.4 容错不靠备份靠重新计算Spark的容错策略也是源于DAG模型的一个巧妙设计。对于一个执行失败的分区最直接的做法是重新运行那个Task。如果重新运行仍然失败Spark会检查程序是否提供了重试机制之外的恢复途径。这里的关键在于血缘关系。当某个分区数据丢失时Spark会用它所在Stage的血缘链从检查点或者初始数据源开始把丢失的分区重新计算出来。这比传统的“对每个中间结果做备份副本”策略节省了大量存储开销。代价是如果血缘链很深重算成本会很高所以实际工程中对于血缘特别长且中间结果很贵的场景我们会手动插入checkpoint把中间结果持久化到可靠存储截断血缘链。4. RDD、DataFrame、DataSet三套API怎么选4.1 三套API的定位差异先梳理一下三套API的核心区别因为很多新手都会在这里纠结。RDD是最底层的抽象它强调函数式编程任何逻辑都可以通过map、filter、flatMap、reduce等算子组合出来。它的优点是灵活几乎能表达任何计算逻辑缺点是运行时自由度太高引擎很难做自动优化。所有优化都必须开发者自己手动来控制。DataFrame是在RDD之上引入Schema信息后的产物它把数据组织成一张带列名的表。有了Schema引擎就能在计算之前对执行计划做深度优化比如谓词下推、列剪枝、常量折叠等。这个优化器就是大家常说的Catalyst。DataSet则是DataFrame的强类型版本它在编译时就能校验类型正确性在某些需要类型安全的场景下比如写一个长期维护的应用价值很大。在Spark 2.x以后DataFrame本质上是DataSet[Row]的特例两者底层共用同一套执行引擎。4.2 Catalyst优化器到底优化了什么Catalyst优化器本质上是一个基于规则的执行计划转换引擎。它把SQL或者DataFrame算子转换成逻辑计划然后应用一系列优化规则。举两个最常见的优化效果。第一个是谓词下推如果你在SQL里写了where过滤条件优化器会尽量把过滤操作“下沉”到数据源读取之后立刻执行这样后续的join和聚合操作处理的数据量就会大幅减少。第二个是列剪枝如果你的查询只需要三列数据优化器会告诉底层的数据源只读取这三列而不是把整行数据都读进来。这些优化在RDD层面是做不到的因为RDD完全不知道每列的名字和类型。这就是为什么官方一直建议优先使用DataFrame和SQL来处理业务逻辑——哪怕你手动对RDD做了很多精细的调优大多时候也比不上优化器自动产出的执行计划。4.3 实际选型建议基于我在多个项目里的实践这里给出一个比较稳妥的选择策略场景推荐API原因常规数据清洗与统计DataFrame / SQL有优化器加持开发效率高复杂ETL逻辑且需要全链路可控DataFrame UDF兼顾灵活性和性能底层算法开发、自定义分区器RDD精细控制执行细节需要类型安全、长期演进的应用DataSet编译期检查类型方便重构重构补充一点不要为了“炫技”而刻意使用RDD。绝大多数业务场景里DataFrame的表现已经足够好而且代码可读性远高于一堆算子串联。RDD的价值更多体现在引擎扩展和底层原语场景。5. 不止批处理Spark生态里的SQL、流、图与机器学习5.1 Spark SQL数据处理最顺手的入口Spark SQL是整个生态的入口之一。它既支持纯SQL也支持编程式API背后是同一个Catalyst优化器。这意味着你可以用一套代码逻辑处理相同语义的两种写法执行计划基本上是一样的。实际使用时Spark SQL之前最典型的落地场景包括把一份数据从文件或者外部数据源读入后注册成临时表然后用标准SQL做复杂的多层嵌套查询。相比直接写RDD算子SQL的表达成本要低得多尤其是遇到多表关联和子查询时可读性优势非常明显。5.2 从Spark Streaming到Structured Streaming流处理这部分早期Spark Streaming是微批处理模式把实时数据流切分成极小的批次按批处理的方式执行延迟一般在秒级。后来推出的Structured Streaming则把流数据抽象成“无界表”每来一条数据就像往表里插入一行然后你可以在上面做连续查询。必须说清楚的是Spark的流处理并不是真正的逐条事件驱动它依然是一个微批模型。在需要毫秒级延迟的实时场景比如金融高频交易Spark并不是最优解。但在秒级延迟能够接受的场景例如系统日志的实时监控、用户行为指标的秒级聚合Structured Streaming凭借“一套代码跑批流”的优势运维成本比维护两套系统低很多。5.3 MLlib与GraphX的实际定位MLlib是Spark上的机器学习库提供了大量经典的分类、回归、聚类、协同过滤算法。GraphX则提供了基于图的计算框架支持PageRank、连通分量等图算法。这里要泼一点冷水这两个库在很多互联网公司里使用率其实没有SQL和流处理那么高。深度学习领域已经被其他更专业的框架占据MLlib的主场更多在于大规模样本上的传统机器学习特征处理和模型训练。图计算则更常被专门的图数据库和向量化图引擎取代。如果你想在简历上写“熟悉Spark全家桶”MLlib和GraphX必须知道它们解决什么问题、适合哪些场景但不必花太长时间深入研究每个算法实现细节。5.4 什么场景适合上Spark全家桶简单概括如果团队的数据处理需求可以统一成“以SQL为中心的离线/准实时加工管道”那么Spark全家桶几乎是成本最低的选项。这通常意味着我们只需要维护一套数据引擎就能覆盖日常报表、临时查询、特征抽取、分钟级事件指标加工等需求整体技术栈非常整洁。反过来如果你需要HTAP能力、毫秒级点查响应或者复杂的事务处理那Spark就不是合适的主角它更适合在数据管道中承担批处理和流式预计算这两层核心角色。6. 部署形态与性能调优里绕不开的坑6.1 四种部署形态如何选Spark支持本地模式local、独立集群模式Standalone、YARN模式以及Kubernetes模式。本地模式一般只用于开发测试和学习Standalone适合小规模集群部署最简单YARN模式在传统大数据平台里非常常见因为它能和已有的HDFS、Hive等组件共用一个资源管理层运维方便Kubernetes模式则是云原生环境下越来越主流的选择动态扩缩容能力更强对基础设施管理有标准化要求的团队更友好。选型的逻辑很简单能少维护一套资源管理系统就少维护一套。如果公司已经有一个大集群在跑HDFS和Hive那么YARN模式是不需要犹豫的选择如果基础设施已经容器化为了统一交付和快速扩缩容优先考虑Kubernetes。6.2 核心参数到底应该怎么配这一节讲调优我认为最重要的一个认知是Spark参数没有“万能最优解”所有最优配置都是针对特定场景的。但如果非要说哪些参数决定了下限那大概率是下面几个首先是spark.executor.memory它决定了每个Executor的JVM堆内存大小。这个参数会直接影响shuffle和缓存溢出的频率以及GC压力。一般建议根据单机总内存和并发Executor数来反推。比如一台机器有64GB内存计划跑2个Executor且预留15%给系统其他进程使用那么每个Executor可以给到24GB左右。其次是spark.executor.cores它决定了每个Executor上的并发Task数。这个参数不能只看CPU核心数还要考虑Task的IO密集程度。一个Executor内部并发Task太多CPU上下文切换开销会明显上升并发太少则浪费核数。常见的经验值是2到4个核心。然后是spark.sql.shuffle.partitions它决定了shuffle后数据的分区数。很多默认配置在生产环境规模下并不合理比如这个参数默认是200但对于特别大或者特别小的数据都需要手动调整。判断方法很简单看每个Task处理的数据量。一般来说单个Task处理的数据量在几百MB到1GB之间比较合理据此反推分区数。最后是spark.default.parallelism在未显式指定分区数时它会作为RDD转换的默认并行度参考值。6.3 数据倾斜、小文件、序列化问题这三个问题几乎是我在真实项目里见得最多的故障源。数据倾斜本质上是某些key的数据量远超其他key导致个别Task压力过大而其他Task早早空闲。最简单的治标手段是给key加随机前缀后分两阶段聚合或者想办法把热点key单独拆分处理。治本手段则需要从业务逻辑层面寻找更好的分片策略。小文件问题则与分区数设置直接有关。如果读取了理论上千个HDFS文件切片但在最终写出时分区数设置非常小生成的文件数量很少但每只很大反过来如果每个批次的分区数量非常多输出时就会产生海量小文件给下游HDFS的NameNode内存和数据读取都带来沉重负担。常用的手段是调整写出阶段的分区数或者先执行一次重分区操作再做输出。序列化方面最常见的坑是使用了Java原生序列化且自定义类没有实现可序列化接口或者是闭包中引用了无法被序列化的对象。除了改用Kryo序列化并注册对应的类我建议在开发阶段多做小规模数据量的本地冒烟测试把这类问题尽早暴露出来。7. 实战复盘一个离线日志统计任务的完整落地过程7.1 业务场景与需求定义为了把前面的知识点串起来分享一个我实际做过的模拟项目某数据平台需要每天对全量用户访问日志做统计核心指标包括各端口的访问热力分布、用户留存分层、异常请求占比并产出结果供下游报表和实时看板使用。这个场景非常适合用Spark SQL来做输入数据是按小时落地的Parquet文件体积在数百GB量级属于典型的“批量大、逻辑不算太重”的场景。下面用一个简化版的核心代码和参数配置来复盘整个过程。7.2 代码实现与参数配置数据读取方面直接读取指定目录下的Parquet文件并以DataFrame注册为临时表。核心统计逻辑需要先对原始事件做去重和清洗再按多维维度做聚合。为了控制单次作业的规模使用传入日期作为参数做分区裁剪。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(daily_log_stat) \ .config(spark.sql.shuffle.partitions, 120) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() input_path /data/logs/events/date2024-05-20 df spark.read.parquet(input_path) df.createOrReplaceTempView(events) sql SELECT channel, platform, COUNT(DISTINCT user_id) AS uv, COUNT(*) AS pv, SUM(IF(is_abnormal 1, 1, 0)) AS abnormal_cnt FROM events WHERE dt 2024-05-20 AND event_type page_view GROUP BY channel, platform result spark.sql(sql) result.write.mode(overwrite).format(parquet).partitionBy(channel).save(/data/result/daily_log_stat)这段代码看起来很简单但实际踩了一个不小的坑COUNT(DISTINCT user_id)在数据量很大的时候会产生严重的shuffle负担尤其是去重基数特别高的维度组合。为了解决这个问题做了一次方案调整把“精确去重”改成了“近似去重”使用approx_count_distinct函数从而大幅降低了shuffle数据量精度控制在可接受范围内。7.3 运行过程中遇到的坑与调整链路第一次全量运行时作业跑了将近40分钟而且其中有一个阶段明显拖尾部分Task执行时间超过15分钟其他Task三分钟就结束了。这是非常典型的节点资源不均加数据倾斜混合问题。排查思路是先在Spark UI里找到拖尾的Stage看它的Task数量和数据量分布。结果发现倾斜的key来自某个数据异常的渠道占整体数据量的约35%。当时没有直接在SQL里强行加盐而是先跟业务方确认了这个渠道的特殊性最后决定在清洗阶段把异常渠道单独分流处理然后再把结果合并。经过这轮调整作业总时长降到了12分钟左右。7.4 调优前后的对比指标调优前调优后总执行时长约38分钟约12分钟最长Task时长16分钟4分钟shuffle数据量约2.1TB约0.6TBExecutor数4040可以看到在资源规模完全没变的前提下仅仅通过优化去重逻辑、拆分倾斜key、合理设置shuffle分区数就把作业耗时压缩到了原来的三分之一。这也是我想反复强调的Spark调优不是堆机器而是真正理解DAG调度和shuffle机制之后把计算尽可能变成“不需要挪数据”的形态。在具体项目里写Spark作业时我还有一个习惯先关注Spark UI上的三个核心指标——Shuffle Read Size、Task执行时间长尾情况、GC时间占比。这三个指标能快速定位绝大多数性能问题。另外预算允许的话开启Spark的Adaptive Query Execution动态执行计划特性它能自动在运行时调整后续Stage的分区数并做广播优化对SQL类作业的提升非常明显。大数据计算这块真正能拉开差距的地方往往不是用了多新的组件而是对DAG调度的每一层细节做到心里有数。Spark发展到今天生态已经非常成熟但底层的那些概念——血缘、依赖、阶段、任务、shuffle、本地性——依然是理解它的全部钥匙。把这套基础打牢了不管以后它怎么演进你都能比较快地看懂新的设计到底改了什么。 
长河君 · 2026/10/14 8:16:04 · 阅读 1.2k Spark核心机制全解析：从DAG调度到实战调优
算上这一篇“开源组件系列”已经写到了第十二篇。前面聊过不少存储和调度相关的组件这篇来把计算引擎这块的核心拼图补上Spark。说实话现在出去面试或者接大数据相关的项目Spark基本上是跳不过去的一环。很多人对Spark的理解停留在“比MapReduce快很多的内存计算框架”但真要问到它为什么快、DAG模型到底解决了什么问题、一个作业从提交到出结果在集群里经历了什么能讲清楚的人其实不多。这篇不打算照着官方文档翻译一遍。我想从“为什么要用DAG来组织计算”这个底层逻辑开始把Spark的作业调度、运行机制、三套核心API的区别、部署形态以及我在实际项目中踩过的调优和排坑经验串起来聊。适合两类人看一类是刚接触Spark、想系统建立认知框架的开发者另一类是已经写过不少Spark代码、但总觉得调度和优化部分隔着一层窗户纸的同学。1. Spark靠什么接过了大数据计算的大旗1.1 在Spark出现之前批处理是怎么跑的要理解Spark的价值得先回到它之前那个时代。早期的大数据批处理主流方案是基于MapReduce模型的。MapReduce把计算过程强行拆成两个阶段Map阶段和Reduce阶段中间隔着一次落盘和网络传输。Map产出的中间结果必须写到本地磁盘等Reduce阶段启动后再去拉取。这种设计最大的问题在于一个复杂的分析任务通常需要串联很多个MapReduce作业前一个作业的输出是后一个作业的输入每一轮都要经过“写磁盘、网络传输、再读磁盘”的过程。单纯从这个流程就能看出来磁盘IO和序列化开销会占据整个作业执行时间相当高的比例。如果是一个需要十几轮迭代的算法比如一些图计算或者机器学习类的任务每一轮都把中间结果刷到磁盘性能会非常难看。这本质上不是MapReduce模型本身算错了而是它对“中间数据”的处理方式太浪费。1.2 DAG把“一步一落盘”变成了“内存里接力”Spark对这件事最核心的改动有两点引入有向无环图DAG来组织计算流程以及把中间结果尽量驻留在内存中。先说DAG。一个数据处理任务从数据源到最终结果中间涉及的各种转换操作比如过滤、映射、聚合、关联这些操作之间的依赖关系天然就能表达成一张图。Spark把这整张DAG交给调度引擎调度引擎对图进行解析找出哪些计算可以放在同一个阶段Stage里串行执行哪些阶段之间存在shuffle依赖必须分开然后以最优的方式把任务分发到集群里的各个节点上去执行。这跟MapReduce那种“机械地拆成固定两段”完全不同DAG的灵活性在于它允许一个阶段的结果直接被下一个阶段消费中间省去了大量落盘。配合内存缓存机制那些需要反复迭代的计算就能在内存里连续接力完成。这也是Spark在迭代式计算和交互式查询场景下比老方案快出几个数量级的根本原因。1.3 内存计算不是越猛越好谈到这里需要纠正一个常见的误解Spark并不是完全不需要磁盘。当内存不足以存放全部中间数据时Spark仍然会把溢出部分写到磁盘也就是常见的spark.shuffle.spill机制。内存计算的意义在于“减少不必要的落盘”而不是“完全消灭落盘”。理解了这点后续做内存参数调优时就不容易走极端。另外有一点很多人在初学阶段容易忽略Spark虽然比老方案快但它不是一个“内存数据库”那样的系统。它的核心依然是一个分布式计算框架只是通过DAG调度和内存复用来大幅缩短了作业的执行时间。搞清楚这层定位对后面我们讨论资源参数配置会很有帮助。2. DAG计算模型的底层拆解2.1 RDDDAG的节点与边Spark的DAG不是凭空产生的它的构建基础是RDD弹性分布式数据集。我们可以把RDD理解为一个“带分区信息的不可变数据集”它记录了数据分布在哪些节点上以及对这个数据集已经施加过哪些转换操作。在Spark里用户的程序基本上就是在不断地创建新的RDD并对其进行转换比如从一个文本文件创建RDD然后map、filter、join每一步操作都会生成一个记录了“我这个数据集是从哪个父数据集、经过什么计算得到的”的RDD。这一系列RDD之间的依赖关系串起来就是一张血缘关系图Lineage Graph。Spark的DAG调度器拿到的正是这张血缘图。这里值得强调的是“不可变”这个特性。RDD一旦创建就不会被修改任何转换操作都生成新的RDD而不是改动原有数据。这个设计带来一个巨大的红利如果某个分区的计算失败或者数据丢失Spark可以沿着血缘关系回溯只重新计算受影响的那个分区而不是把整个作业从头跑一遍。这就是容错机制的基础。2.2 窄依赖与宽依赖划分Stage的标尺光有图还不够调度器要执行这张图必须搞清楚哪些计算能“就地串联”哪些计算必须“停下来等待数据重排”。这就需要区分两类依赖窄依赖是指父RDD的每个分区最多被子RDD的一个分区使用。典型的操作就是map、filter、union。窄依赖场景下父分区的计算和子分区的计算可以在同一个任务里连续完成不需要跨节点传数据效率极高。宽依赖是指父RDD的每个分区可能被子RDD的多个分区使用典型的就是groupByKey、reduceByKey、join这一类会触发shuffle的操作。宽依赖存在时必须等父阶段的全部任务执行完然后按照key重新分区和传输数据才能开始下一阶段的计算。DAG调度器划分Stage的规则很简单粗暴从后往前回溯遇到宽依赖就把依赖链切断切出来的每一段就是一个Stage。换句话说Stage的边界就是shuffle发生的位置。2.3 一张图看懂执行顺序下面用一个小场景把这几个概念串起来。假设有这样的一个计算逻辑读取两份数据集一份做过滤并聚合另一份做映射然后两者做关联最后排序输出。在实际执行时Spark先生成DAG然后DAG调度器发现关联操作join是一个宽依赖因此它被设为Stage边界关联之前的所有操作会被压缩到前一个Stage里关联之后直到排序输出则属于后一个Stage。同一个Stage内部的任务之间不需要数据重排可以尽可能地利用内存和CPU流水线式执行。这个流程实际对应的执行计划是DAG操作是否触发shuffleStage归属读取源数据否Stage 0过滤、映射、局部聚合否Stage 0关联操作是Stage 0与Stage 1的边界排序与结果输出是Stage 12.4 DAG相比两阶段模型的优势到这里可以做个明确的对比了。MapReduce模型中每个业务逻辑都被强行套进Map和Reduce两个阶段里复杂任务需要手动拆成多个作业串行执行层层落盘。DAG模型中计算图在调度层是全局可见的调度器能自动识别哪些计算可以合并执行哪些必须分开然后把尽可能多的计算放进同一个Stage内完成。这种“全局视角”带来的直接收益是对于多步骤任务作业数量从原来的“每个MapReduce作业都要完整走一遍提交、调度、执行、落盘的流程”变成了“一次提交内部自动划分阶段”。作业粒度和调度开销都大幅下降尤其适合那些需要几十步转换的复杂ETL逻辑。3. 一次Spark作业在集群里的完整旅程3.1 Driver、Executor和集群管理器的三角关系理解DAG调度模型之后有必要看看一个作业提交后到底经过了哪些环节。Spark采用经典的主从架构Driver负责解析代码、构建DAG、划分Stage、调度TaskExecutor负责真正执行Task并在内存里缓存数据。集群管理器可以是自带的Standalone模式也可以对接YARN或者Kubernetes负责资源的分配和回收。有个概念容易混淆Driver和集群管理器各自管什么。Driver管的是“计算任务的编排”它知道要执行哪些Task、每个Task的数据在哪集群管理器管的是“计算资源的分配”它只知道有多少个Executor可以启动每个Executor给多少CPU和内存至于Task在Executor内部怎么跑它不关心。3.2 作业、阶段、任务三层调度结构Spark的调度是典型的三层模型。一个完整的作业Job是用户提交的一个动作比如一次count操作或者一个SQL查询一个作业会被拆成多个阶段Stage阶段之间以宽依赖为界每个阶段对应多个任务TaskTask的数量取决于分区数每个分区由一个Task负责处理。这三层关系值得多说明一下。Task是实际在Executor上运行的单元一个Executor默认情况下可以并发运行多个Task具体取决于分配的CPU核心数。比如分配了4个核心那这个Executor上就能同时跑4个Task。分区数和并行度直接挂钩这个数设置太小会导致资源利用率不足设置太大又会增加调度开销。3.3 数据本地性让计算主动靠近数据Spark调度还有一个容易被忽略但影响很大的策略数据本地性。它的基本原则是“能不传数据就不传数据尽量让Task到数据所在的节点上去执行”。因为网络传输的代价通常比在节点本地读数据要高得多尤其对于TB级别的大数据集数据的移动会造成巨大的IO和网络开销。本地性由高到低分为几个等级进程本地性PROCESS_LOCAL数据就在同一个Executor里、节点本地性NODE_LOCAL数据在同一个节点的其他Executor里、机架本地性RACK_LOCAL数据在同一个机架的其他节点上等。调度器会先尽量满足高等级本地性如果等待超时还拿不到对应位置的资源会降级执行从其他节点拉取数据。3.4 容错不靠备份靠重新计算Spark的容错策略也是源于DAG模型的一个巧妙设计。对于一个执行失败的分区最直接的做法是重新运行那个Task。如果重新运行仍然失败Spark会检查程序是否提供了重试机制之外的恢复途径。这里的关键在于血缘关系。当某个分区数据丢失时Spark会用它所在Stage的血缘链从检查点或者初始数据源开始把丢失的分区重新计算出来。这比传统的“对每个中间结果做备份副本”策略节省了大量存储开销。代价是如果血缘链很深重算成本会很高所以实际工程中对于血缘特别长且中间结果很贵的场景我们会手动插入checkpoint把中间结果持久化到可靠存储截断血缘链。4. RDD、DataFrame、DataSet三套API怎么选4.1 三套API的定位差异先梳理一下三套API的核心区别因为很多新手都会在这里纠结。RDD是最底层的抽象它强调函数式编程任何逻辑都可以通过map、filter、flatMap、reduce等算子组合出来。它的优点是灵活几乎能表达任何计算逻辑缺点是运行时自由度太高引擎很难做自动优化。所有优化都必须开发者自己手动来控制。DataFrame是在RDD之上引入Schema信息后的产物它把数据组织成一张带列名的表。有了Schema引擎就能在计算之前对执行计划做深度优化比如谓词下推、列剪枝、常量折叠等。这个优化器就是大家常说的Catalyst。DataSet则是DataFrame的强类型版本它在编译时就能校验类型正确性在某些需要类型安全的场景下比如写一个长期维护的应用价值很大。在Spark 2.x以后DataFrame本质上是DataSet[Row]的特例两者底层共用同一套执行引擎。4.2 Catalyst优化器到底优化了什么Catalyst优化器本质上是一个基于规则的执行计划转换引擎。它把SQL或者DataFrame算子转换成逻辑计划然后应用一系列优化规则。举两个最常见的优化效果。第一个是谓词下推如果你在SQL里写了where过滤条件优化器会尽量把过滤操作“下沉”到数据源读取之后立刻执行这样后续的join和聚合操作处理的数据量就会大幅减少。第二个是列剪枝如果你的查询只需要三列数据优化器会告诉底层的数据源只读取这三列而不是把整行数据都读进来。这些优化在RDD层面是做不到的因为RDD完全不知道每列的名字和类型。这就是为什么官方一直建议优先使用DataFrame和SQL来处理业务逻辑——哪怕你手动对RDD做了很多精细的调优大多时候也比不上优化器自动产出的执行计划。4.3 实际选型建议基于我在多个项目里的实践这里给出一个比较稳妥的选择策略场景推荐API原因常规数据清洗与统计DataFrame / SQL有优化器加持开发效率高复杂ETL逻辑且需要全链路可控DataFrame UDF兼顾灵活性和性能底层算法开发、自定义分区器RDD精细控制执行细节需要类型安全、长期演进的应用DataSet编译期检查类型方便重构重构补充一点不要为了“炫技”而刻意使用RDD。绝大多数业务场景里DataFrame的表现已经足够好而且代码可读性远高于一堆算子串联。RDD的价值更多体现在引擎扩展和底层原语场景。5. 不止批处理Spark生态里的SQL、流、图与机器学习5.1 Spark SQL数据处理最顺手的入口Spark SQL是整个生态的入口之一。它既支持纯SQL也支持编程式API背后是同一个Catalyst优化器。这意味着你可以用一套代码逻辑处理相同语义的两种写法执行计划基本上是一样的。实际使用时Spark SQL之前最典型的落地场景包括把一份数据从文件或者外部数据源读入后注册成临时表然后用标准SQL做复杂的多层嵌套查询。相比直接写RDD算子SQL的表达成本要低得多尤其是遇到多表关联和子查询时可读性优势非常明显。5.2 从Spark Streaming到Structured Streaming流处理这部分早期Spark Streaming是微批处理模式把实时数据流切分成极小的批次按批处理的方式执行延迟一般在秒级。后来推出的Structured Streaming则把流数据抽象成“无界表”每来一条数据就像往表里插入一行然后你可以在上面做连续查询。必须说清楚的是Spark的流处理并不是真正的逐条事件驱动它依然是一个微批模型。在需要毫秒级延迟的实时场景比如金融高频交易Spark并不是最优解。但在秒级延迟能够接受的场景例如系统日志的实时监控、用户行为指标的秒级聚合Structured Streaming凭借“一套代码跑批流”的优势运维成本比维护两套系统低很多。5.3 MLlib与GraphX的实际定位MLlib是Spark上的机器学习库提供了大量经典的分类、回归、聚类、协同过滤算法。GraphX则提供了基于图的计算框架支持PageRank、连通分量等图算法。这里要泼一点冷水这两个库在很多互联网公司里使用率其实没有SQL和流处理那么高。深度学习领域已经被其他更专业的框架占据MLlib的主场更多在于大规模样本上的传统机器学习特征处理和模型训练。图计算则更常被专门的图数据库和向量化图引擎取代。如果你想在简历上写“熟悉Spark全家桶”MLlib和GraphX必须知道它们解决什么问题、适合哪些场景但不必花太长时间深入研究每个算法实现细节。5.4 什么场景适合上Spark全家桶简单概括如果团队的数据处理需求可以统一成“以SQL为中心的离线/准实时加工管道”那么Spark全家桶几乎是成本最低的选项。这通常意味着我们只需要维护一套数据引擎就能覆盖日常报表、临时查询、特征抽取、分钟级事件指标加工等需求整体技术栈非常整洁。反过来如果你需要HTAP能力、毫秒级点查响应或者复杂的事务处理那Spark就不是合适的主角它更适合在数据管道中承担批处理和流式预计算这两层核心角色。6. 部署形态与性能调优里绕不开的坑6.1 四种部署形态如何选Spark支持本地模式local、独立集群模式Standalone、YARN模式以及Kubernetes模式。本地模式一般只用于开发测试和学习Standalone适合小规模集群部署最简单YARN模式在传统大数据平台里非常常见因为它能和已有的HDFS、Hive等组件共用一个资源管理层运维方便Kubernetes模式则是云原生环境下越来越主流的选择动态扩缩容能力更强对基础设施管理有标准化要求的团队更友好。选型的逻辑很简单能少维护一套资源管理系统就少维护一套。如果公司已经有一个大集群在跑HDFS和Hive那么YARN模式是不需要犹豫的选择如果基础设施已经容器化为了统一交付和快速扩缩容优先考虑Kubernetes。6.2 核心参数到底应该怎么配这一节讲调优我认为最重要的一个认知是Spark参数没有“万能最优解”所有最优配置都是针对特定场景的。但如果非要说哪些参数决定了下限那大概率是下面几个首先是spark.executor.memory它决定了每个Executor的JVM堆内存大小。这个参数会直接影响shuffle和缓存溢出的频率以及GC压力。一般建议根据单机总内存和并发Executor数来反推。比如一台机器有64GB内存计划跑2个Executor且预留15%给系统其他进程使用那么每个Executor可以给到24GB左右。其次是spark.executor.cores它决定了每个Executor上的并发Task数。这个参数不能只看CPU核心数还要考虑Task的IO密集程度。一个Executor内部并发Task太多CPU上下文切换开销会明显上升并发太少则浪费核数。常见的经验值是2到4个核心。然后是spark.sql.shuffle.partitions它决定了shuffle后数据的分区数。很多默认配置在生产环境规模下并不合理比如这个参数默认是200但对于特别大或者特别小的数据都需要手动调整。判断方法很简单看每个Task处理的数据量。一般来说单个Task处理的数据量在几百MB到1GB之间比较合理据此反推分区数。最后是spark.default.parallelism在未显式指定分区数时它会作为RDD转换的默认并行度参考值。6.3 数据倾斜、小文件、序列化问题这三个问题几乎是我在真实项目里见得最多的故障源。数据倾斜本质上是某些key的数据量远超其他key导致个别Task压力过大而其他Task早早空闲。最简单的治标手段是给key加随机前缀后分两阶段聚合或者想办法把热点key单独拆分处理。治本手段则需要从业务逻辑层面寻找更好的分片策略。小文件问题则与分区数设置直接有关。如果读取了理论上千个HDFS文件切片但在最终写出时分区数设置非常小生成的文件数量很少但每只很大反过来如果每个批次的分区数量非常多输出时就会产生海量小文件给下游HDFS的NameNode内存和数据读取都带来沉重负担。常用的手段是调整写出阶段的分区数或者先执行一次重分区操作再做输出。序列化方面最常见的坑是使用了Java原生序列化且自定义类没有实现可序列化接口或者是闭包中引用了无法被序列化的对象。除了改用Kryo序列化并注册对应的类我建议在开发阶段多做小规模数据量的本地冒烟测试把这类问题尽早暴露出来。7. 实战复盘一个离线日志统计任务的完整落地过程7.1 业务场景与需求定义为了把前面的知识点串起来分享一个我实际做过的模拟项目某数据平台需要每天对全量用户访问日志做统计核心指标包括各端口的访问热力分布、用户留存分层、异常请求占比并产出结果供下游报表和实时看板使用。这个场景非常适合用Spark SQL来做输入数据是按小时落地的Parquet文件体积在数百GB量级属于典型的“批量大、逻辑不算太重”的场景。下面用一个简化版的核心代码和参数配置来复盘整个过程。7.2 代码实现与参数配置数据读取方面直接读取指定目录下的Parquet文件并以DataFrame注册为临时表。核心统计逻辑需要先对原始事件做去重和清洗再按多维维度做聚合。为了控制单次作业的规模使用传入日期作为参数做分区裁剪。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(daily_log_stat) \ .config(spark.sql.shuffle.partitions, 120) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() input_path /data/logs/events/date2024-05-20 df spark.read.parquet(input_path) df.createOrReplaceTempView(events) sql SELECT channel, platform, COUNT(DISTINCT user_id) AS uv, COUNT(*) AS pv, SUM(IF(is_abnormal 1, 1, 0)) AS abnormal_cnt FROM events WHERE dt 2024-05-20 AND event_type page_view GROUP BY channel, platform result spark.sql(sql) result.write.mode(overwrite).format(parquet).partitionBy(channel).save(/data/result/daily_log_stat)这段代码看起来很简单但实际踩了一个不小的坑COUNT(DISTINCT user_id)在数据量很大的时候会产生严重的shuffle负担尤其是去重基数特别高的维度组合。为了解决这个问题做了一次方案调整把“精确去重”改成了“近似去重”使用approx_count_distinct函数从而大幅降低了shuffle数据量精度控制在可接受范围内。7.3 运行过程中遇到的坑与调整链路第一次全量运行时作业跑了将近40分钟而且其中有一个阶段明显拖尾部分Task执行时间超过15分钟其他Task三分钟就结束了。这是非常典型的节点资源不均加数据倾斜混合问题。排查思路是先在Spark UI里找到拖尾的Stage看它的Task数量和数据量分布。结果发现倾斜的key来自某个数据异常的渠道占整体数据量的约35%。当时没有直接在SQL里强行加盐而是先跟业务方确认了这个渠道的特殊性最后决定在清洗阶段把异常渠道单独分流处理然后再把结果合并。经过这轮调整作业总时长降到了12分钟左右。7.4 调优前后的对比指标调优前调优后总执行时长约38分钟约12分钟最长Task时长16分钟4分钟shuffle数据量约2.1TB约0.6TBExecutor数4040可以看到在资源规模完全没变的前提下仅仅通过优化去重逻辑、拆分倾斜key、合理设置shuffle分区数就把作业耗时压缩到了原来的三分之一。这也是我想反复强调的Spark调优不是堆机器而是真正理解DAG调度和shuffle机制之后把计算尽可能变成“不需要挪数据”的形态。在具体项目里写Spark作业时我还有一个习惯先关注Spark UI上的三个核心指标——Shuffle Read Size、Task执行时间长尾情况、GC时间占比。这三个指标能快速定位绝大多数性能问题。另外预算允许的话开启Spark的Adaptive Query Execution动态执行计划特性它能自动在运行时调整后续Stage的分区数并做广播优化对SQL类作业的提升非常明显。大数据计算这块真正能拉开差距的地方往往不是用了多新的组件而是对DAG调度的每一层细节做到心里有数。Spark发展到今天生态已经非常成熟但底层的那些概念——血缘、依赖、阶段、任务、shuffle、本地性——依然是理解它的全部钥匙。把这套基础打牢了不管以后它怎么演进你都能比较快地看懂新的设计到底改了什么。