Spark 核心之 Driver 原理剖析

摘要如果把 Spark 应用比作一个人Driver 就是它的大脑。从 spark-submit 敲下回车的那一刻起SparkContext 初始化、DAGScheduler 切分 Stage、TaskScheduler 分发 Task、SchedulerBackend 与集群通信、SparkEnv 管理运行时环境——所有这些都在 Driver 内部精密协作。本文从 Driver 内部架构全景、SparkContext 初始化链路、三大调度器协作模型、SparkEnv 七大组件、Driver 生命周期六个维度配合 1 张原创深色架构图和完整源码级分析带你彻底看清 Driver 的内部世界。关键词Spark Driver, SparkContext, DAGScheduler, TaskScheduler, SchedulerBackend, SparkEnv, BlockManager, Driver 生命周期一、开篇Driver 是什么先回答一个面试高频题“Spark Driver 到底做了什么”答案不是一句话能说完的。Driver 是 Spark 应用的总控制器它承载了至少以下七大职责#职责核心组件1解析用户代码 → 构建 DAGDAGScheduler2将 DAG 切分为 StageDAGScheduler3将 Stage 拆分为 Task 并分发TaskSchedulerImpl4与集群通信申请/释放资源SchedulerBackend5管理运行时环境内存/序列化/ShuffleSparkEnv6事件监听与 Web UILiveListenerBus SparkUI7Executor 心跳监控HeartbeatReceiver二、Driver 内部架构全景图2.1 三大组件群┌─────────────────────────────────────────────────┐ │ SparkContext │ │ ┌─────────────┐ ┌─────────────┐ ┌──────────┐│ │ │核心调度组件 │ │ SparkEnv │ │ 监控/事件 ││ │ │DAGScheduler │ │BlockManager │ │LiveListen ││ │ │TaskScheduler│ │ShuffleMgr │ │SparkUI ││ │ │SchedulerBknd│ │MemoryMgr │ │MetricsSys ││ │ └─────────────┘ └─────────────┘ └──────────┘│ └─────────────────────────────────────────────────┘三、SparkContext 初始化链路这是 Driver 启动最核心的代码路径。// 源码SparkContext.scala (简化版初始化链路)classSparkContext(config:SparkConf)extendsLogging{// Step 1: 创建 SparkEnv运行时环境privatevar_env:SparkEnv_ _envSparkEnv.createDriverEnv(conf,isLocal,listenerBus,...)// Step 2: 创建元数据追踪器_applicationId_env.conf.get(spark.app.id)_dagSchedulernewDAGScheduler(this)// Step 3: 创建 TaskScheduler SchedulerBackendval(sched,ts)SparkContext.createTaskScheduler(this,master,deployMode)_schedulerBackendsched _taskSchedulerts// Step 4: DAGScheduler 绑定 TaskScheduler_dagSchedulernewDAGScheduler(this)_taskScheduler.start()// Step 5: 启动心跳接收器_heartbeatReceiverenv.rpcEnv.setupEndpoint(HeartbeatReceiver.ENDPOINT_NAME,newHeartbeatReceiver(this))// Step 6: 注册 SparkListener 启动 WebUIsetupAndStartListenerBus()_uiSparkUI.create(conf,listenerBus,_env,...)}四、三大调度器协作模型 这是 Driver 最核心的调度链路。用户代码 (Action) │ ▼ DAGScheduler.handleJobSubmitted() │ ① 回溯 RDD 依赖 → 创建 ResultStage │ ② getMissingParentStages() → 递归构建 ShuffleMapStage │ ③ submitMissingTasks() → 为每个 Partition 创建 Task ▼ TaskScheduler.submitTasks(taskSet) │ ④ TaskSetManager 封装 → 数据本地性排序 │ ⑤ reviveOffers() → 通知 Backend 有 Task 可调度 ▼ SchedulerBackend.reviveOffers() │ ⑥ makeOffers() → 匹配空闲 Executor 与 Task │ ⑦ launchTasks() → 序列化 Task 发送给 Executor ▼ Executor (远程 JVM) ⑧ 反序列化 → 执行 → 序列化结果 → StatusUpdate4.1 DAGSchedulerStage 切分核心// 源码核心逻辑privatedefsubmitStage(stage:Stage):Unit{valmissinggetMissingParentStages(stage).sortBy(_.id)if(missing.isEmpty){submitMissingTasks(stage,jobId.get)}else{for(parent-missing)submitStage(parent)}}privatedefgetMissingParentStages(stage:Stage):List[Stage]{stage.rdd.dependencies.flatMap{caseshufDep:ShuffleDependency[_,_,_]getOrCreateShuffleMapStage(shufDep,stage.firstJobId)case_Nil// NarrowDep 不切分}.toList}规则遇到 ShuffleDependency 即切分 Stage。4.2 TaskSchedulerImpl数据本地性// 数据本地性优先级PROCESS_LOCALNODE_LOCALRACK_LOCALANY// 每个级别等待 spark.locality.wait (默认 3s)4.3 SchedulerBackend集群通信适配器实现通信目标StandaloneSchedulerBackendSpark Master (Netty RPC)YarnSchedulerBackendYARN AM → RM (Hadoop RPC)KubernetesClusterSchedulerBackendK8s API Server (HTTP REST)五、SparkEnv运行时环境七大组件// 源码SparkEnv.createDriverEnv()valblockManagernewBlockManager(...)valbroadcastManagernewBroadcastManager(...)valmapOutputTrackernewMapOutputTrackerMaster(...)valshuffleManagerSortShuffleManager(conf)valmemoryManagerUnifiedMemoryManager(conf,...)valserializernewJavaSerializer(conf)// or KryoSerializervalclosureSerializernewJavaSerializer(conf)组件职责BlockManagerRDD 缓存Memory Disk、Shuffle 数据存储MapOutputTracker追踪 Shuffle Map 输出位置Master/WorkerShuffleManagerSortShuffleManager 管理 Shuffle 写/读MemoryManagerUnifiedMemoryManager执行 存储统一内存池SerializerTask 序列化/反序列化BroadcastManagerTorrentBroadcast 分布式广播RpcEnvNettyRpcEnvDriver ↔ Executor 通信基础设施六、Driver 完整生命周期Phase 1: spark-submit → main() → new SparkContext() ├── 创建 SparkEnv运行时环境 ├── 创建 DAGScheduler TaskScheduler SchedulerBackend ├── 向 Master/RM 注册申请 Executor └── 启动 HeartbeatReceiver SparkUI Phase 2: Action 触发 → DAG 调度 ├── DAGScheduler.handleJobSubmitted() ├── Stage 切分 Task 生成 ├── TaskScheduler 分发 Task └── Executor 执行 StatusUpdate 回传 Phase 3: 监控与运维 ├── LiveListenerBus 推送事件 ├── SparkUI :4040 实时监控 └── HeartbeatReceiver 心跳检测 Phase 4: sc.stop() → 优雅退出 ├── 通知 SchedulerBackend 停止 ├── Kill 全部 Executor ├── 向 Master/RM 注销 └── 释放 SparkEnv 资源七、Driver 配置调优spark-submit\--driver-memory 4G\# Driver JVM 堆内存--driver-cores2\# Driver 可用核心--confspark.driver.maxResultSize2G\# collect() 结果上限--confspark.driver.extraJavaOptions-XX:UseG1GC\--confspark.driver.extraClassPath/path/to/extra.jar\--confspark.driver.supervisetrue\# Standalone Cluster 专属my-app.jar八、总结要点总结Driver 本质用户 main() 运行的 JVM 进程SparkContext 即 Driver 入口三大调度器DAGScheduler → TaskScheduler → SchedulerBackend 逐层下发SparkEnv7 大组件提供序列化、Shuffle、内存、存储等运行时能力生命周期初始化 → 调度循环 → 监控 → 优雅退出金句Executor 是 Spark 的四肢Driver 是 Spark 的大脑——DAGScheduler 思考如何拆分TaskScheduler 决定派给谁SchedulerBackend 负责怎么送。作者starzy | AI Data Engineer / 大数据技术实践者博客blog.starzy.cn | GitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践