大数据技术之_19_Spark学习_06_Spark 源码解析小结

Posted 2021-11-26 chenmingjun

========== 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 模式启动流程 ==========

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

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

1、Driver 提交流程：用户通过 spark-submit 将 jar 包和相对应的参数提交给 spark 框架，内部实现是通过 ClientEndpoint 向 Master 发送了 RequestSubmitDriver 消息，Master 获取消息之后通过 Worker 进行 Lau nchDriver 操作。
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 部署模式 ==========