spark源码解析总结

========== Spark 通信架构 ==========

spark源码解析总结
1、spark 一开始使用 akka 作为网络通信框架,spark 2.X 版本以后完全抛弃 akka,而使用 netty 作为新的网络通信框架。
最主要原因:spark 对 akka 没有维护,需要 akka 更新,spark 的发展受到了 akka 的牵制,akka 版本之间无法通信,即 akka 兼容性问题。
2、RpcEnv:RPC 上下文环境,每个 Rpc 端点运行时依赖的上下文环境称之为 RpcEnv。类似于 SparkContext,默认由 NettyRpcEnv 实现,由 NettyRpcEnvFactory 创建 RpcEnv。
3、RpcEndpoint:RPC 端点,Spark 针对于每个节点(Client/Master/Worker)都称之一个 Rpc 端点且都实现 RpcEndpoint 接口,内部根据不同端点的需求,设计不同的消息和不同的业务处理,如果需要发送(询问)则调用 Dispatcher。代理是 RpcEndpointRef。
4、Dispatcher:消息分发器,针对于 RPC 端点需要发送消息或者从远程 RPC 接收到的消息,分发至对应的指令收件箱/发件箱。
5、Inbox:指令消息收件箱,一个本地端点对应一个收件箱,Dispatcher 在每次向 Inbox 存入消息时,都将对应 EndpointData 加入内部待 Receiver Queue 中。
6、OutBox:指令消息发件箱,一个远程端点对应一个发件箱,当消息放入 Outbox 后,紧接着将消息通过 TransportClient 发送出去。
7、TransportClient:Netty 通信客户端,主要负责将相对应的 OutBox 中的数据发送给远程 TransportServer。
8、TransportServer:Netty 通信服务端,主要用于接收远程 RpcEndpoint 发送过来的消息,并把消息传送给 Dispatcher。

========== Spark 脚本解析 ==========
1、start-slave.sh 用于启动 slave 节点,最终启动的类是 org.apache.spark.deploy.worker.Worker 类。
2、start-master.sh 用于启动 master 节点,最终启动的类是 org.apache.spark.deploy.master.Master 类。
3、spark-submit 和 spark-shell 最终都会调用 spark-class 脚本,通过 spark-class 脚本启动相对应的入口类。

========== Spark standalone 模式启动流程 ==========

spark源码解析总结
1、Master 和 Worker 都继承了 RpcEndpoint 类,成为了具体的消息发送与接收端点,整个应用是利用 Actor 模型实现的异步消息通信架构。
2、Master 节点在启动的时候的主要任务是创建了通信架构中的 RpcEnv,并注册了 Master 成为端点。
3、Worker 节点在启动的时候的主要任务是创建了通信架构中的 RpcEnv,并注册了 Worker 成为端点,并且获取了 Master 端点的代理,通过端点代理向 Master 发送消息。
4、Worker 节点在启动的时候执行 onStar 方法,向 Master 进行了注册。

========== Spark 应用提交流程 ==========

spark源码解析总结
1、Driver 提交流程:用户通过 spark-submit 将 jar 包和相对应的参数提交给 spark 框架,内部实现是通过 ClientEndpoint 向 Master 发送了 RequestSubmitDriver 消息,Master 获取消息之后通过 Worker 进行 LaunchDriver 操作。
2、Driver 的进程启动:主要通过 Worker 节点的 DriverRunner 来启动整个的 Driver 进程。
3、注册 Application:Driver 进程在启动之后,通过 SparkContext 的初始化操作,创建了对应的 StandaloneSchedulerBackend,实现了向 Master 进行当前应用的注册。
4、启动 Executor 进程:当 Driver 向 Master 进行注册之后,Master 通过 scheduler() 方法来对当前的 App 进行 Executor 的分配,实现上是通过 Worker 的 ExecutorRunner 来进行 Executor 的创建和运行。
5、启动 Task 运行:当 Driver 收到所有的 Executor 资源后,通过 RDD 的 action 操作,触发 SparkContext.runJob 方法,进而调用 Dagscheduler() 方法进行当前 DAG 的运行。通过向 Executor 发送 LaunchTask 消息来启动 Executor 上的任务运行。
6、Task 运行完成:当 Executor 运任务完成之后,会通知 Driver 当前任务的运行状态,然后执行任务 或者退出整个应用。

========== Spark shuffle 过程 ==========
MapReduce Shuffle 过程
1、在 spill(刷写)阶段,数据直接写入到 kvbuffer 数据缓冲器中。会写两种类型的数据。一种是 kvmeta 数据,用于存放分区信息、索引信息;另一种是 (k, v) 对类型的数据,是实际的数据。
2、会以一个起点反向来写,即当遇到 spill 进程启动的时候,写入点会重新进行选择。

Hash Shuffle 过程
1、未优化版本,每一个 task 任务都会根据 reduce 任务的个数创建对应数量的 bucket,bucket 其实就是写入缓冲区,每一个 bucket 都会存入一个文件,这个文件叫做 blockfile。最大的缺点是:产生的文件过多。
2、在优化版本中,主要通过 consolidation 这个参数进行优化,实现了 ShuffleFileGroup 的概念,不同批次的 task 任务可以复用最终写入的文件,来整体减少文件的数量。

Sort Shuffle 过程
1、Sort Shuffle 整个过程的实现和 MapReduce Shuffle 过程类似。
2、Bypass 机制:Hash Shuffle 在 reduce 的数量比价少的时候性能要比 Sort Shuffle 要高,所以如果你的 reduce 的数量少于 Bypass 定义的数值的时候,Sort Shuffle 在 task 任务写出的时候会采用 Hash 方式,而不会采用 ApplyOnlyMap 以及排序的方法。

========== Spark 内存管理与分配 ==========
1、内存分配模式:静态分配和统一分配。静态分配就是固定大小分配,统一分配就是存储区和 Shuffle 区可以动态占用。
2、有几种内存配置模式:
  (1)other 区,一般占用 20% 的内存区域,主要是用于代码运行以及相关数据的运行。
  (2)Execution 区,这个区域一般占用 20% 的内存区域,主要用于 Shuffle 过程的内存消耗,通过 spark.shuffle.memeoryFaction 参数进行控制。
  (3)Storage 区,这个区域主要用于 RDD 的缓存,通过 spark.storage.memeoryFaction 参数进行控制。
3、spark 目前支持堆内内存和堆外内存,堆外内存主要用于存储序列化后的二进制数据。

========== Spark 部署模式 ==========