Spark on YARN 加速启动

// org.apache.spark.deploy.yarn.Client
private def createContainerLaunchContext {
    ... 
    val appStagingDirPath = new Path(appStagingBaseDir, getAppStagingDir(appId))
    ...
    val localResources = prepareLocalResources(appStagingDirPath, pySparkArchives)
    ...
}
// org.apache.spark.deploy.yarn.Client
def prepareLocalResources {
    ...
    
    def distribute {
      val trimmedPath = path.trim()
      val localURI = Utils.resolveURI(trimmedPath)
      if (localURI.getScheme != LOCAL_SCHEME) {
        if (addDistributedUri(localURI)) {
          val localPath = getQualifiedLocalPath(localURI, hadoopConf)
          val linkname = targetDir.map(_ + "/").getOrElse("") +
           destName.orElse(Option(localURI.getFragment())).getOrElse(localPath.getName())
          val destPath = copyFileToRemote(destDir, localPath, replication, symlinkCache)
          val destFs = FileSystem.get(destPath.toUri(), hadoopConf)
          distCacheMgr.addResource(
            destFs, hadoopConf, destPath, localResources, resType, linkname, statCache,
            appMasterOnly = appMasterOnly)
          (false, linkname)
        } else {
          (false, null)
        }
      } else {
        (true, trimmedPath)
      }
    }
    ...
    /**
     * Add Spark to the cache. There are two settings that control what files to add to the cache:
     * - if a Spark archive is defined, use the archive. The archive is expected to contain
     *   jar files at its root directory.
     * - if a list of jars is provided, filter the non-local ones, resolve globs, and
     *   add the found files to the cache.
     *
     * Note that the archive cannot be a "local" URI. If none of the above settings are found,
     * then upload all files found in $SPARK_HOME/jars.
     */
    val sparkArchive = sparkConf.get(SPARK_ARCHIVE)
    if (sparkArchive.isDefined) {
      val archive = sparkArchive.get
      require(!isLocalUri(archive), s"${SPARK_ARCHIVE.key} cannot be a local URI.")
      distribute(Utils.resolveURI(archive).toString,
        resType = LocalResourceType.ARCHIVE,
        destName = Some(LOCALIZED_LIB_DIR))
    }else {
      sparkConf.get(SPARK_JARS) match {
      	case Some(jars) =>{
            // 操作类似 SPARK_ARCHIVE，把 SPARK_JARS 上传；两者设置一个即可
            ... 
        }
       case None =>
          // No configuration, so fall back to uploading local jar files.
          logWarning(s"Neither ${SPARK_JARS.key} nor ${SPARK_ARCHIVE.key} is set, falling back " + "to uploading libraries under SPARK_HOME.")
          // 把 spark/jars 所有 jar 上传
      }
       
    ...
}
 private[yarn] def copyFileToRemote {
        val destFs = destDir.getFileSystem(hadoopConf)
        val srcFs = srcPath.getFileSystem(hadoopConf)
        var destPath = srcPath
        if (force || !compareFs(srcFs, destFs) || "file".equals(srcFs.getScheme)) {
            destPath = new Path(destDir, destName.getOrElse(srcPath.getName()))
            logInfo(s"Uploading resource $srcPath -> $destPath")
            FileUtil.copy(srcFs, srcPath, destFs, destPath, false, hadoopConf)
            destFs.setReplication(destPath, replication)
            destFs.setPermission(destPath, new FsPermission(APP_FILE_PERMISSION))
        } else {
            logInfo(s"Source and destination file systems are the same. Not copying $srcPath")
        }
        ...
}

    // This code forces the archive to be copied, so that unit tests pass (since in that case both
    // file systems are the same and the archive wouldn't normally be copied). In most (all?)
    // deployments, the archive would be copied anyway, since it's a temp file in the local file
    // system.
    val remoteConfArchivePath = new Path(destDir, LOCALIZED_CONF_ARCHIVE)
    val remoteFs = FileSystem.get(remoteConfArchivePath.toUri(), hadoopConf)
    sparkConf.set(CACHED_CONF_ARCHIVE, remoteConfArchivePath.toString())

    val localConfArchive = new Path(createConfArchive().toURI())
    copyFileToRemote(destDir, localConfArchive, replication, symlinkCache, force = true,
      destName = Some(LOCALIZED_CONF_ARCHIVE))