[SPARK-3634] [PySpark] User's module should take precedence over system modules

python/pyspark/context.py

@@ -171,7 +171,7 @@ def _do_init(self, master, appName, sparkHome, pyFiles, environment, batchSize,
 
         SparkFiles._sc = self
         root_dir = SparkFiles.getRootDirectory()
-        sys.path.append(root_dir)
+        sys.path.insert(1, root_dir)
 
         # Deploy any code dependencies specified in the constructor
         self._python_includes = list()
@@ -183,10 +183,9 @@ def _do_init(self, master, appName, sparkHome, pyFiles, environment, batchSize,
         for path in self._conf.get("spark.submit.pyFiles", "").split(","):
             if path != "":
                 (dirname, filename) = os.path.split(path)
-                self._python_includes.append(filename)
-                sys.path.append(path)
-                if dirname not in sys.path:
-                    sys.path.append(dirname)
+                if filename.lower().endswith("zip") or filename.lower().endswith("egg"):
+                    self._python_includes.append(filename)
+                    sys.path.insert(1, os.path.join(SparkFiles.getRootDirectory(), filename))
 
         # Create a temporary directory inside spark.local.dir:
         local_dir = self._jvm.org.apache.spark.util.Utils.getLocalDir(self._jsc.sc().conf())
@@ -667,7 +666,7 @@ def addPyFile(self, path):
         if filename.endswith('.zip') or filename.endswith('.ZIP') or filename.endswith('.egg'):
             self._python_includes.append(filename)
             # for tests in local mode
-            sys.path.append(os.path.join(SparkFiles.getRootDirectory(), filename))
+            sys.path.insert(1, os.path.join(SparkFiles.getRootDirectory(), filename))
 
     def setCheckpointDir(self, dirName):
         """

python/pyspark/tests.py

@@ -323,6 +323,18 @@ def func():
         from userlib import UserClass
         self.assertEqual("Hello World from inside a package!", UserClass().hello())
 
+    def test_overwrite_system_module(self):
+        self.sc.addPyFile(os.path.join(SPARK_HOME, "python/test_support/SimpleHTTPServer.py"))
+
+        import SimpleHTTPServer
+        self.assertEqual("My Server", SimpleHTTPServer.__name__)
+
+        def func(x):
+            import SimpleHTTPServer
+            return SimpleHTTPServer.__name__
+
+        self.assertEqual(["My Server"], self.sc.parallelize(range(1)).map(func).collect())
+
 
 class TestRDDFunctions(PySparkTestCase):
 

python/pyspark/worker.py

@@ -43,6 +43,13 @@ def report_times(outfile, boot, init, finish):
     write_long(1000 * finish, outfile)
 
 
+def add_path(path):
+    # worker can be used, so donot add path multiple times
+    if path not in sys.path:
+        # overwrite system packages
+        sys.path.insert(1, path)
+
+
 def main(infile, outfile):
     try:
         boot_time = time.time()
@@ -61,11 +68,11 @@ def main(infile, outfile):
         SparkFiles._is_running_on_worker = True
 
         # fetch names of includes (*.zip and *.egg files) and construct PYTHONPATH
-        sys.path.append(spark_files_dir)  # *.py files that were added will be copied here
+        add_path(spark_files_dir)  # *.py files that were added will be copied here
         num_python_includes = read_int(infile)
         for _ in range(num_python_includes):
             filename = utf8_deserializer.loads(infile)
-            sys.path.append(os.path.join(spark_files_dir, filename))
+            add_path(os.path.join(spark_files_dir, filename))
 
         # fetch names and values of broadcast variables
         num_broadcast_variables = read_int(infile)

python/test_support/SimpleHTTPServer.py

@@ -0,0 +1,22 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Used to test override standard SimpleHTTPServer module.
+"""
+
+__name__ = "My Server"