Please enable JavaScript.

Coggle requires JavaScript to display documents.

Spark (核心概念 (RDD 是不可变的， lazy级别的，粗粒度的数据集合，包含了一个或者多个数据分片，即partition …

- - - - DAGScheduler 负责高层调度，划分stage并生成程序运行的有向无环图
      - TaskScheduler 负责具体的stage内部task的调度和容错
      - Job: (正在执行的叫ActiveJob) 是TOP-Level的工作单位，每个action算子都会触发一次Job，一个Job可能包含一个或者多个stage
      - Stage: 是用来计算中间结果的Tasksets. Tasksets中的task逻辑对于同一个RDD内的不同partition都一样。Stage在shuffle的地方产生，此时下一个stage要用上一个stage的全部数据，所以要等上一个stage全部执行完毕才能开始。
        Stage分为两种: ShuffleMapStage和ResuleStage, 最后一个stage是ResultStage,其他的都是ShuffleMapStage。
        ShuffleMapStage会产生中间结果，以文件的形式保存在集群中，stage经常被不同的Job共享，前提是Job重用了同一个RDD。
      - Task 任务执行的工作单位，每个task会被发送到一个节点上，一个task对应RDD的一个partition
    - - （1）StandaloneScheulerBackend负责集群资源的管理和调度，这个是从作业角度考虑的，注册给master的时候，Master给我们分配资源，资源从executor本身转过来向StandaloneScheulerBackend注册，这是从作业调度的角度来考虑的，不是从整个集群考虑，整个集群是master来管理计算资源。
        StandaloneScheulerBackend有三大核心功能：
        
        负责与Master链接，注册当前程序RegisterWithMaster。
        
        接收集群中为当前应用程序分配的计算资源executor的注册并管理executors.
        
        负责发送task到具体的executor执行。
        
        SchedulerBackend包含多个子类：
        a. LocalSchedulerBackend
        b. CoarseGrainedSchedulerBackend
        c. StandaloneScheudlerBackend
        d. MesosCoarseGrainedSchedulerBackend
        e. YarnScheudlerBackend
      - （2）DAGscheduler负责高层调度： job中的stage划分，数据本地性等
        它为每一个job计算DAG，跟踪RDDS以及stage输出结果进行物化，并找到一个最小的计划去执行job，然后提交stages中的taskSet到底层的TaskScheduler调度器，task scheduler再提交task到集群运行。taskset包含完全独立的任务，基于集群上已存在的数据行，如果这个数据不可用，获取数据可能失败。
      - （3）TaskSchedulerImple负责具体stage内部底层调度：每个task的调度，task的容错。
      - （4）MapOutputTrackerMaster 负责shuffle中数据的输出和读取的管理。
        shuffle的时候将数据写到本地，下一个stage要用上一个stage的数据，因此写数据的时候就需要告诉Driver中MapOutputTrackerMaster具体写到哪里，下一个stage读取数据的时候也要访问Driver的MapOutputTrackerMaster 获取数据的具体的位置。
  - - - （1）数据分布存储在多台机器上，事实上，每个RDD的数据都已Block的形式存储与多台机器上，每个executor会启动一个BlockManagerSlave并管理一部分Block；
        Block的元数据由Driver节点上的BlockManagerMaster保存，BlockManagerSlave生成Block后向BlockManagerMaster注册该Block，BlockManagerMaster管理RDD和Block的关系，当RDD不需要存储的时候，将向BlockManagerSlave发送指令删除相应的Block。
      - （2）BlockManager管理RDD的物理分区，每个Block就是节点上对应的一个数据块，可以存储在内存或者磁盘上。而RDD中的partition是一个逻辑数据块，对应相应的物理块Block。
        本质上，一个RDD在代码中相当于数据的一个元数据结构，存储着数据分区及其逻辑结构映射关系，存储着RDD之前的依赖转换关系。
        BlockManager在每个节点上运行和管理Block(Driver&Executors)，它提供一个接口检索本地和远程的存储变量，比如memory, disk, off-heap。使用 BlockManager必须要初始化。
      - （3）一个RDD在代码中相当于数据的一个元数据结构（一个RDD就是一组分区），存储着数据分区以及Block, Node等的映射关系，以及其他的元数据信息，存储着RDD之间的依赖转换关系。
        分区是一个逻辑概念，tansfromation前后的新旧分区在物理上有可能是同一个内存存储。
    - - （1）分区列表(a list of partitions)
        RDD是被分区的，每一个分区都会被计算任务task处理，分区数决定并行计算数量，RDD的并行度默认从父RDD传给子RDD，默认情况下，一个HDFS上的数据分片就是一个partition， RDD分片数量决定了并行计算的力度，可以在创建RDD的收指定，如果不只从分区的数量，当RDD从集合创建时，则默认分区数量为该程序所分配到的CPU核数（每个core可以承载2-4个partitions），如果从HDFS文件创建，默认为文件的Block数。
      - （2）每个分区都有一个计算函数，a function for compution each split. spark的RDD的计算函数是以分片为基本单位的，每个RDD都会实现compute函数，对具体的分片进行计算，RDD中的分片是并行的，所以是分布式并行计算。
        spark根据宽依赖划分stage，stage内部通过pipeline操作，通过BlockManager获取相关的数据，因为具体的split要从外界读取数据，也要把具体的数据写入到外界，所以用了一个管理器，具体的split都会映射成BlockManager的Block，而具体的split会被函数处理，函数处理的形式是以任务的形式进行的。
      - （3）依赖其他的RDD列表
        依赖关系：宽依赖和窄依赖
        父RDDpartition --> 子RDD partition
        多 --> 一窄依赖
        多 --> 多宽依赖
        spark中的宽依赖指的是生成的RDD的每一个partition都依赖于父RDD的所有的partition，宽依赖的定性操作有: groupByKey, sortByKey等，宽依赖意味着shuffle操作，这是spark划分stage边界的依据，spark中宽依赖支持两种shuffle manager：HashShuffleManager和SortShuffleManager,前者是基于Hash shuffle的机制，后置是基于排序shuffle的机制。HashShuffleManager已经废弃。
      - （4）key-value 数据类型的RDD分区器，控制分区策略和分区数。每个key-value形式的RDD都有partitoner属性，它决定了RDD如何分区。
        Partition的个数还决定每个stage的task个数。
      - （5）每个分区都有一个优先位置列表。它会存储每个partition的优先位置，对于一个HDFS文件来说，就是每个partition块的位置。
        观察运行 spark 集群的控制台会发现 Spark 的具体计算，具体分片前，它已经清楚地知道任务发生在什么节点上，也就是说，任务本身是计算层面的、代码层面的，代码发生运算之前已经知道它要运算的数据在什么地方，有具体节点的信息。这就符合大数据中数据不动代码动的特点。数据不动代码动的最高境界是数据就在当前节点的内存中。这时有可能是 memoy 级别或 Alluxio 级别的， Spark 本身在进行任务调度时候，会尽可能将任务分配到处理数据的数据块所在的具体位置。据 Spark 的 RDD.Scala 源码函数 getPreferredLocations 可知，每次计算都符合完美的数据本地性
    - - （1）RDD的多个partition分别由不同的 task处理。task分为两类：ShuffleMapTask 和 resultTask
      - （2）task是计算运行在集群上的基本计算单位。一个task负责处理RDD的一个partition，一个RDD的多个partition分别由不同的task处理。
        Spark计算中对于每一个stage分配的task的数目都是基于这个stage中最后一个RDD的partition的个数来决定的。
        Task运行在executor上，而executor位于CoarseGrainedExecutorBackend的JVM进程中。
        Spark job 中，根据task所处的stage的位置，将task分为两类：ShuffleMapTask 和 resultTask；
        ShuffleMapTask指的是task所处的stage不是最后一个stage，也就是stage的计算结果还没有输出，二十通过shuffle叫个下一个stage是用。
        resultTask指的是task所处的stage是DAG中最后一个stage，也就是这个stage计算结果需要输出操作，计算到此结束。
      - CoarseGrainedExecutorBackend执行task
        
        （1）首先反序列化 TaskDescription
        （2）Executor通过launchTask执行task
        （3）Executor的launchTask方法会创建一个TaskRunner实列在threadPool来运行具体的Task
        （4）Eexecutor.run中调用反序列化后的Task.run方法来执行任务，并获取执行结果。 task.run方法调用了runTask方法，而runTask方法是一个抽象的方法，runTask方法内部会调用RDD的iterator()方法，该方法就是针对当前task对应的partition进行计算的关键所在，在处理的方法内部会迭代partition的元素，并交给我们自己定义的function来处理。
    - - （1）容错原理：由于宽依赖实质上是指父RDD的一个分区会对应一个子RDD的多个分区，这种情况下出现部分计算结果丢失，单一计算丢失的数据无法达到效果，便采用重新计算该步骤中的所有数据，从而导致计算数据重复；
        对于窄依赖而言，窄依赖实质指的是父RDD的一个分区最多被子RDD的一个分区使用，此情况下出现部分计算错误，由于计算结果的数据只依赖于父RDD的相关数据，所以不需要重新计算所有数据，只需要重新计算出错的部分即可。
      - RDD容错的三大核心层面
        
        （1）调度层（包含DAG生成和Task重算两大核心）
        
        （1）DAG生成
        stage输出失败，上层调度器DAGScheduler会进行重试。
        通过DAGScheduler的resubmitFailedStages方法进行重试。
        
        （2）Task计算层
        spark计算的过程中，计算内部某个task任务失败了，底层调度去TaskSetManager会对这个task进行若干次的重试。默认4次。
        TaskSetManager中handleFailedTaask负责进行重试。
        
        （2）RDD lineage 血统层容错
        
        spark中RDD采用高度受限的分布式共享内存，且新的RDD的产生只能通过其他的RDD上的批量操作来创建，依赖于RDD的lineage为核心的容错处理。
        Spark RDD实现基于lineage的容错机制，基于RDD的各项transformation构成了compute chain，在计算部分结果部分丢失的时候根据lineage重新恢复计算。
        
        在窄依赖中，在子RDD的分区丢失，要重新算父RDD分区，父RDD相应的分区的所有数据都是子RDD的分区数据，并存在冗余计算。
        
        在宽依赖的情况下，丢失了一个子RDD分区，重算每个父RDD的每个分区的所有数据并不是都给丢失的子RDD分区用的，会有一部分的数据相当于对应的是为丢失的子RDD的分区数据，这样会产生冗余计算和巨大的性能浪费。
        
        （3）checkpoint层容错
        
        spark的checkpoint通过将RDD写入Disk做检查点，是spark lineage容错的辅助，lineage过程会造成容错的成本很高，这时在中间阶段做检查点容错，乳沟之后有节点出现问题而丢失分区，从做checkpoint的RDD开始重做lineage，就会减少开销。
        
        checkpoint适用两种情况
        
        （1）DAG中的lineage过长，如果重新计算，开销太大，如PageRank
        
        （2）尤其适合在宽依赖上做checkpoint，这个时候可以避免为lineage重新计算而带来的冗余计算。
  - - - （1） spark中，DAG生成的流程关键在于【回溯】，在程序提交以后，高层调度器(DAGScheduler)将所有的RDD看成一个stage，然后对此stage进行从后往前的回溯，遇到shuffle就断开，遇到窄依赖，则归并到同一个stage中。等到所有的步骤回溯完成，就生成了一个DAG图。
      - （2）spark stage 根据RDD途中的shuffle的边界来创建，如果RDD的操作是窄依赖，如map, filter ，在每个stages中将一些列tasks组合成流水线执行。但是如果是宽依赖，shuffle依赖需要多个stages（上一个stage进行map输出到文件，下一个stage读取数据文件），每个stage依赖于其他的stage，其中进行多个算子操作。算子操作在各种类型的RDDS的RDD.compute()中实际执行。
      - （3）缓存跟踪： DAGScheduler记录了那些RDDS被缓存了，避免重复计算，以及记录了shuffle map stages 已经生成的输出文件，避免在map端重新计算。
      - （4）数据本地化 DAGScheduler基于RDDs的数据本地性，缓存位置，或者shuffle数据在stage中运行每一个任务的task。
        清理：当依赖与他们运行的作业完成时，所有数据将被清理，防止在长时间运行的应用程序中内存泄漏。