From 90b38b9ed395bca7c1a872a1ceeac536e8196550 Mon Sep 17 00:00:00 2001
From: hyukjinkwon <gurwls223@apache.org>
Date: Fri, 25 May 2018 12:43:53 +0800
Subject: [PATCH 1/2] Add .py files correctly into PythonRunner in submit with
 client mode

---
 .../apache/spark/deploy/PythonRunner.scala    | 29 ++++++++++++++++++-
 .../spark/deploy/yarn/YarnClusterSuite.scala  | 15 ++++------
 2 files changed, 33 insertions(+), 11 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala b/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
index 1b7e031ee0678..bfa7ad6d5c43b 100644
--- a/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
@@ -19,6 +19,7 @@ package org.apache.spark.deploy
 
 import java.io.File
 import java.net.{InetAddress, URI}
+import java.nio.file.Files
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
@@ -48,7 +49,7 @@ object PythonRunner {
 
     // Format python file paths before adding them to the PYTHONPATH
     val formattedPythonFile = formatPath(pythonFile)
-    val formattedPyFiles = formatPaths(pyFiles)
+    val formattedPyFiles = resolvePyFiles(formatPaths(pyFiles))
 
     // Launch a Py4J gateway server for the process to connect to; this will let it see our
     // Java system properties and such
@@ -153,4 +154,30 @@ object PythonRunner {
       .map { p => formatPath(p, testWindows) }
   }
 
+  /**
+   * Resolves the ".py" files. ".py" file should not be added as is because PYTHONPATH does
+   * not expect a file. This method creates a temporary directory and puts the ".py" files
+   * if exist in the given paths.
+   */
+  private def resolvePyFiles(pyFiles: Array[String]): Array[String] = {
+    lazy val dest = Utils.createTempDir(namePrefix = "localPyFiles")
+    pyFiles.flatMap { pyFile =>
+      // In case of client with submit, the python paths should be set before context
+      // initialization because the context initialization can be done later.
+      // We will copy the local ".py" files because ".py" file shouldn't be added
+      // alone but its parent directory in PYTHONPATH. See SPARK-24384.
+      if (pyFile.endsWith(".py")) {
+        val source = new File(pyFile)
+        if (source.exists() && source.canRead) {
+          Files.copy(source.toPath, new File(dest, source.getName).toPath)
+          Some(dest.getAbsolutePath)
+        } else {
+          // Don't have to add it if it doesn't exist or isn't readable.
+          None
+        }
+      } else {
+        Some(pyFile)
+      }
+    }.distinct
+  }
 }
diff --git a/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala b/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala
index 59b0f29e37d84..3b78b88de778d 100644
--- a/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala
+++ b/resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala
@@ -271,16 +271,11 @@ class YarnClusterSuite extends BaseYarnClusterSuite {
       "PYSPARK_ARCHIVES_PATH" -> pythonPath.map("local:" + _).mkString(File.pathSeparator),
       "PYTHONPATH" -> pythonPath.mkString(File.pathSeparator)) ++ extraEnv
 
-    val moduleDir =
-      if (clientMode) {
-        // In client-mode, .py files added with --py-files are not visible in the driver.
-        // This is something that the launcher library would have to handle.
-        tempDir
-      } else {
-        val subdir = new File(tempDir, "pyModules")
-        subdir.mkdir()
-        subdir
-      }
+    val moduleDir = {
+      val subdir = new File(tempDir, "pyModules")
+      subdir.mkdir()
+      subdir
+    }
     val pyModule = new File(moduleDir, "mod1.py")
     Files.write(TEST_PYMODULE, pyModule, StandardCharsets.UTF_8)
 

From f015e0d587c8d9f8cd359fecc325a19362a59c55 Mon Sep 17 00:00:00 2001
From: hyukjinkwon <gurwls223@apache.org>
Date: Tue, 29 May 2018 11:03:21 +0800
Subject: [PATCH 2/2] Add isFile sanity check too

---
 core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala b/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
index bfa7ad6d5c43b..ccb30e205ca40 100644
--- a/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/PythonRunner.scala
@@ -168,7 +168,7 @@ object PythonRunner {
       // alone but its parent directory in PYTHONPATH. See SPARK-24384.
       if (pyFile.endsWith(".py")) {
         val source = new File(pyFile)
-        if (source.exists() && source.canRead) {
+        if (source.exists() && source.isFile && source.canRead) {
           Files.copy(source.toPath, new File(dest, source.getName).toPath)
           Some(dest.getAbsolutePath)
         } else {