vlambda博客
学习文章列表

spark.driver.extraLibraryPathspark.executor.extraClassPath

 <property>
 <description>Classpath for typical applications.</description>
 <name>yarn.application.classpath</name>
 <value>/opt/modules/spark-2.1.2/jars/*,$HADOOP_CONF_DIR,$HADOOP_COMMON_HOME/share/hadoop/common/*,$HADOOP_COMMON_HOME/share/hadoop/common/lib/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/*,
$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*,$HADOOP_YARN_HOME/share/hadoop/yarn/*,$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*
 </value>
 </property>

spark.driver.userClassPathFirstspark.executor.userClassPathFirst

  private[spark] val addedJars = new ConcurrentHashMap[String, Long]().asScala

// SPARK-21563 make a copy of the jars/files so they are consistent across the TaskSetprivate val addedJars = HashMap[String, Long](sched.sc.addedJars.toSeq: _*)private val addedFiles = HashMap[String, Long](sched.sc.addedFiles.toSeq: _*)

 new TaskDescription( taskId, attemptNum, execId, taskName, index, task.partitionId, addedFiles, addedJars, task.localProperties, serializedTask) }

updateDependencies(taskDescription.addedFiles, taskDescription.addedJars)

/** * Download any missing dependencies if we receive a new set of files and JARs from the * SparkContext. Also adds any new JARs we fetched to the class loader.*/private def updateDependencies(newFiles: Map[String, Long], newJars: Map[String, Long]) { lazy val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf) synchronized { // Fetch missing dependencies for ((name, timestamp) <- newFiles if currentFiles.getOrElse(name, -1L) < timestamp) { logInfo("Fetching " + name + " with timestamp " + timestamp) // Fetch file with useCache mode, close cache for local mode. Utils.fetchFile(name, new File(SparkFiles.getRootDirectory()), conf, env.securityManager, hadoopConf, timestamp, useCache = !isLocal) currentFiles(name) = timestamp } for ((name, timestamp) <- newJars) { val localName = new URI(name).getPath.split("/").last val currentTimeStamp = currentJars.get(name) .orElse(currentJars.get(localName)) .getOrElse(-1L) if (currentTimeStamp < timestamp) { logInfo("Fetching " + name + " with timestamp " + timestamp) // Fetch file with useCache mode, close cache for local mode. Utils.fetchFile(name, new File(SparkFiles.getRootDirectory()), conf, env.securityManager, hadoopConf, timestamp, useCache = !isLocal) currentJars(name) = timestamp // Add it to our class loader val url = new File(SparkFiles.getRootDirectory(), localName).toURI.toURL if (!urlClassLoader.getURLs().contains(url)) { logInfo("Adding " + url + " to class loader") urlClassLoader.addURL(url) } } } }}

 // Create our ClassLoader // do this after SparkEnv creation so can access the SecurityManager private val urlClassLoader = createClassLoader() private val replClassLoader = addReplClassLoaderIfNeeded(urlClassLoader)

 /** * Create a ClassLoader for use in tasks, adding any JARs specified by the user or any classes * created by the interpreter to the search path */ private def createClassLoader(): MutableURLClassLoader = { // Bootstrap the list of jars with the user class path. val now = System.currentTimeMillis() userClassPath.foreach { url => currentJars(url.getPath().split("/").last) = now }
 val currentLoader = Utils.getContextOrSparkClassLoader
 // For each of the jars in the jarSet, add them to the class loader. // We assume each of the files has already been fetched. val urls = userClassPath.toArray ++ currentJars.keySet.map { uri => new File(uri.split("/").last).toURI.toURL } if (userClassPathFirst) { new ChildFirstURLClassLoader(urls, currentLoader) } else { new MutableURLClassLoader(urls, currentLoader) } }

/** * A mutable class loader that gives preference to its own URLs over the parent class loader * when loading classes and resources. */public class ChildFirstURLClassLoader extends MutableURLClassLoader {
 static { ClassLoader.registerAsParallelCapable(); }
 private ParentClassLoader parent;
 public ChildFirstURLClassLoader(URL[] urls, ClassLoader parent) { super(urls, null); this.parent = new ParentClassLoader(parent); }
 @Override public Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException { try { return super.loadClass(name, resolve); } catch (ClassNotFoundException cnf) { return parent.loadClass(name, resolve); } }
 @Override public Enumeration<URL> getResources(String name) throws IOException { ArrayList<URL> urls = Collections.list(super.getResources(name)); urls.addAll(Collections.list(parent.getResources(name))); return Collections.enumeration(urls); }
 @Override public URL getResource(String name) { URL url = super.getResource(name); if (url != null) { return url; } else { return parent.getResource(name); } }}

 @Override public Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException { try { return super.loadClass(name, resolve); } catch (ClassNotFoundException cnf) { return parent.loadClass(name, resolve); } }