画布 1
Spark Shuffle优化
  添加Metrics指标
    ShuffleReadMetrics
      incSerialReadTime
      incHDFSReadTime
      incHDFSBlocksCount
      incHDFSBlocksBytes
    ShuffleWriteMetrics
      incTotalWriteTime
      incShuffleHDFSWriteTime
      这两个在三个writer里，没有全部使用
    TaskMetrics
      incShuffleSpillTime
    OutputMetrics
      incWriteTime
        SparkHadoopWriter
  增加的配置参数
    spark.shuffle.hdfs.enabled
      是否开启Spark Shuffle HDFS
    spark.shuffle.hdfs.replication
    spark.shuffle.hdfs.rootDir
    spark.hadoop.io.file.buffer.size
    spark.shuffle.hdfs.fileDel
      作业结束后，是否删除hdfs shuffle数据
  新增HDFSManager
    BlockManager
      hdfsManager 成员
      stop
      initialize
    HdfsManager 类
      appId
  shuffleService
    ExternalShuffleBlockHandler
      handleMessage
        RegisterExecutor
        OpenBlocks
          注册流(StreamId)
            clientID
            blocks的iterator
          返回client: StreamHandle
            StreamId
            blocksNum
    ExternalShuffleBlockResolver
      levelDB维护executor⽬目录信息
      管理理shuffle数据
  reducer
    BlockStoreShuffleReader
      解压缩/反序列化
        combine
          order
    ShuffleBlockFetcherIterator
      1. 划分本地远程数据
splitLocalRemoteBlocks
      2. 获取远程数据(异步)
fetchUpToMaxBytes
        2.1 筛选符合条件的Request，不符合则defer
          - 数据量不能超过48M
          - 请求Request数不能超过Int.MaxValue
          - 同一个地址请求Block数不超过Int.MaxValue
        2.2 发送请求
          2.2.1 请求数据小于Long.MaxValue，直接取到内存中
          2.2.2 否则落盘
        2.3 返回成功，取得的block放入queue中
        2.4 返回失败，FetchFailedException
          去HDFS取数据
      3.获取本地数据
fetchLocalBlocks
        userHDFS时，失败则直接抛IO异常
          一般本地磁盘拿不到可能机器磁盘有问题，即使读hdfs也避免不了后续因为机器磁盘的问题下线，task fail会重新调度别的机器跑
        否则，加入FailureFetchResult到结果队列里
      外部调用next方法
        Success
          A. 从queue中取出获得的block
          B. 创建block数据流
            userHDFS时，失败则直接抛IO异常
              一般本地磁盘拿不到可能机器磁盘有问题，即使读hdfs也避免不了后续因为机器磁盘的问题下线，task fail会重新调度别的机器跑
          C. 检查block是否存在坏块(SPARK-27562)
            存在即FetchFailedException
          D. 将block变成输出流
        Failure
          throwFetchFailedException
  driver
      DAGScheduler.scala
      TaskSetManager
        executorLost(execId: String, host: String, reason: ExecutorLossReason)
      MapOutputTracker.scala
        removeOutputsOnExecutor
    DAGScheduler.scala
      handleWorkerRemoved
        executorLost，添加hdfs.enable判断，判断是否 file lost
      handleExecutorLost
        没有使用hdfs，则shuffle 文件全丢失
      handleTaskCompletion
case FetchFailed
        MapOutputTrackerMaster
          removeOutputsOnHost(host: String)
removeOutputsOnExecutor(execId: String)
unregisterMapOutput(shuffleId: Int, mapId: Int, bmAddress: BlockManagerId)
    driver端停止时删除hdfs数据
      BlockManagerMaster
        stop
  mapper端
    ShuffleWriter 写入
      SortShuffleWriter
      BypassMergeSortShuffleWriter
      UnsafeShuffleWriter
      添加HdfsManager，在write方法时，调用其API写入hdfs