var writer: ShuffleWriter[Any, Any] = null    try {      val manager = SparkEnv.get.shuffleManager      writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)      //這里就是ShuffleMapTask類的runTask()方法中對(duì)應(yīng)的代碼調(diào)用      writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])      writer.stop(success = true).get    } catch {      ...................    }

final def iterator(split: Partition, context: TaskContext): Iterator[T] = {    //判斷下如果StorageLevel.NONE這說明RDD,之前肯定是進(jìn)行了持久化    //getOrCompute中會(huì)通過CacheManager獲取之前持久化的數(shù)據(jù)    if (storageLevel != StorageLevel.NONE) {      getOrCompute(split, context)      //如果沒有進(jìn)行過持久化，就需要通過父RDD定義的算子去獲取數(shù)據(jù)      //注意這里如果有CheckPoint，會(huì)通過CheckPoint獲取，checkPoint獲取不到才去重新計(jì)算      } else {      computeOrReadCheckpoint(split, context)    }  }

 private[spark] def getOrCompute(partition: Partition, context: TaskContext): Iterator[T] = {    val blockId = RDDBlockId(id, partition.index)    var readCachedBlock = true    //CacheManger這里是通過BlockManager獲取持久化數(shù)據(jù)，    //如果獲取成功直接返回，如果獲取失敗，調(diào)用computeOrReadCheckpoint進(jìn)行計(jì)算     //內(nèi)存數(shù)據(jù)為啥會(huì)丟失？ 之前我們知道內(nèi)存中的數(shù)據(jù)如果空間不夠的話，同時(shí)如果指定可以將數(shù)據(jù)緩存到磁盤，會(huì)溢寫到磁盤，   //如果未指定溢寫到磁盤，這些數(shù)據(jù)就會(huì)丟失掉 就需要重新計(jì)算    SparkEnv.get.blockManager.getOrElseUpdate(blockId, storageLevel, elementClassTag, () => {      readCachedBlock = false      //獲取不到重新計(jì)算,這里要注意，代碼執(zhí)行到這里說明這個(gè)RDD肯定是經(jīng)過持久化的      //這里計(jì)算出數(shù)據(jù)后，會(huì)在getOrElseUpdate里面通過makeIterator參數(shù)對(duì)數(shù)據(jù)進(jìn)行重新持久化（這里理解的不太透徹）      computeOrReadCheckpoint(partition, context)    }) match {      case Left(blockResult) =>        if (readCachedBlock) {          val existingMetrics = context.taskMetrics().inputMetrics          existingMetrics.incBytesRead(blockResult.bytes)          new InterruptibleIterator[T](context, blockResult.data.asInstanceOf[Iterator[T]]) {            override def next(): T = {              existingMetrics.incRecordsRead(1)              delegate.next()            }          }        } else {          new InterruptibleIterator(context, blockResult.data.asInstanceOf[Iterator[T]])        }      case Right(iter) =>        new InterruptibleIterator(context, iter.asInstanceOf[Iterator[T]])    }  }

//這里會(huì)調(diào)用get()方法從本地或者遠(yuǎn)程獲取block數(shù)據(jù)，直接返回//如果沒有讀取到數(shù)據(jù)就需要重新計(jì)算數(shù)據(jù)，由于代碼執(zhí)行到這里，rdd肯定是經(jīng)過持久化的//這里計(jì)算出數(shù)據(jù)后，通過makeIterator迭代器，重新進(jìn)行持久化（這里理解的不太透徹） def getOrElseUpdate[T](      blockId: BlockId,      level: StorageLevel,      classTag: ClassTag[T],      makeIterator: () => Iterator[T]): Either[BlockResult, Iterator[T]] = {    // Attempt to read the block from local or remote storage. If it's present, then we don't need    // to go through the local-get-or-put path.    //這里會(huì)調(diào)用get()方法從本地或者遠(yuǎn)程獲取block數(shù)據(jù)，直接返回    get[T](blockId)(classTag) match {      case Some(block) =>        return Left(block)      case _ =>        // Need to compute the block.    }    //這里的處理意思是：對(duì)于本地遠(yuǎn)程沒有獲取到數(shù)據(jù)，然后computeOrReadCheckpoint重新計(jì)算的數(shù)據(jù)    //由于RDD是持久化的，原來的持久化數(shù)據(jù)可能丟了，這里根據(jù)持久化級(jí)別重新進(jìn)行數(shù)據(jù)的持久化    //這里代碼有點(diǎn)不太好理解 要結(jié)合上面2中第12-14行代碼 一起理解    doPutIterator(blockId, makeIterator, level, classTag, keepReadLock = true) match {      case None =>        // doPut() didn't hand work back to us, so the block already existed or was successfully        // stored. Therefore, we now hold a read lock on the block.        val blockResult = getLocalValues(blockId).getOrElse {          // Since we held a read lock between the doPut() and get() calls, the block should not          // have been evicted, so get() not returning the block indicates some internal error.          releaseLock(blockId)          throw new SparkException(s"get() failed for block $blockId even though we held a lock")        }        // We already hold a read lock on the block from the doPut() call and getLocalValues()        // acquires the lock again, so we need to call releaseLock() here so that the net number        // of lock acquisitions is 1 (since the caller will only call release() once).        releaseLock(blockId)        Left(blockResult)      case Some(iter) =>        // The put failed, likely because the data was too large to fit in memory and could not be        // dropped to disk. Therefore, we need to pass the input iterator back to the caller so        // that they can decide what to do with the values (e.g. process them without caching).       Right(iter)    }  }

private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] =  {  //如果設(shè)置了CheckPoint，從Checkpoint中獲取數(shù)據(jù)  //這里CheckPoint相關(guān)的知識(shí)，先不講解，后面有篇文章單獨(dú)講解    if (isCheckpointedAndMaterialized) {          firstParent[T].iterator(split, context)    } else {    //如果數(shù)據(jù)沒有進(jìn)行過Checkpoint，這里只能重新計(jì)算一次    //這里就是根據(jù)自己的rdd算子重新計(jì)算      compute(split, context)    }  }