update code and add comments

Author: lianhuiwang

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -328,8 +328,12 @@ object SparkSubmit {
       }
     }
 
-    if (args.isPython && System.getenv("PYSPARK_ARCHIVES_PATH") != null) {
-      args.files = mergeFileLists(args.files, System.getenv("PYSPARK_ARCHIVES_PATH"))
+    // In yarn mode for a python app, if PYSPARK_ARCHIVES_PATH is in the user environment
+    // add pyspark archives to files that can be distributed with the job
+    if (args.isPython && clusterManager == YARN){
+      sys.env.get("PYSPARK_ARCHIVES_PATH").map { archives =>
+        args.files = mergeFileLists(args.files, Utils.resolveURIs(archives))
+      }
     }
 
     // If we're running a R app, set the main class to our specific R runner

yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -326,10 +326,15 @@ private[spark] class Client(
     distCacheMgr.setDistFilesEnv(env)
     distCacheMgr.setDistArchivesEnv(env)
 
-    if (System.getenv("PYSPARK_ARCHIVES_PATH") != null) {
-      val pythonPath = System.getenv("PYSPARK_ARCHIVES_PATH").split(",").map(
+    // If PYSPARK_ARCHIVES_PATH is in the user environment, set PYTHONPATH to be passed
+    // on to the ApplicationMaster and the executors.
+    sys.env.get("PYSPARK_ARCHIVES_PATH").map { archives =>
+      // archives will be distributed to each machine's working directory, so strip the
+      // path prefix
+      val pythonPath = archives.split(",").map(
         p => (new Path(p)).getName).mkString(":")
       env("PYTHONPATH") = pythonPath
+      sparkConf.setExecutorEnv("PYTHONPATH", pythonPath)
     }
 
     // Pick up any environment variables for the AM provided through spark.yarn.appMasterEnv.*

yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala

@@ -285,10 +285,6 @@ class ExecutorRunnable(
       YarnSparkHadoopUtil.addPathToEnvironment(env, key, value)
     }
 
-    if (System.getenv("PYTHONPATH") != null) {
-      env("PYTHONPATH") = System.getenv("PYTHONPATH")
-    }
-
     // Keep this for backwards compatibility but users should move to the config
     sys.env.get("SPARK_YARN_USER_ENV").foreach { userEnvs =>
       YarnSparkHadoopUtil.setEnvFromInputString(env, userEnvs)