[SPARK-3772] Allow ipython to be used by Pyspark workers; IPython fixes:

- Fix the remaining uses of the '-u' flag, which IPython doesn't support.
- Change PYSPARK_PYTHON_OPTS to PYSPARK_DRIVER_PYTHON_OPTS, so that the old
  name is reserved in case we ever want to allow the worker Python options
  to be customized.

Author: JoshRosen

bin/pyspark

@@ -50,21 +50,36 @@ fi
 
 . "$FWDIR"/bin/load-spark-env.sh
 
-# Figure out which Python executable to use
-if [[ -z "$PYSPARK_PYTHON" ]]; then
-  if [[ "$IPYTHON" = "1" || -n "$IPYTHON_OPTS" ]]; then
-    # for backward compatibility
-    PYSPARK_PYTHON="ipython"
-  else
-    PYSPARK_PYTHON="python"
-  fi
+# In Spark <= 1.1, setting IPYTHON=1 would cause the driver to be launched using the `ipython`
+# executable, while the worker would still be launched using PYSPARK_PYTHON.  This allowed users to
+# launch IPython notebooks with PySpark without requiring IPython to be installed on the workers.
+# Unfortunately, this approach had a few drabacks:
+#
+#   - It wasn't easy to use a custom IPython executable (SPARK-3265).
+#   - There was a risk that the `ipython` and `PYSPARK_PYTHON` executables might run different
+#     Python versions (e.g. 2.6 on driver and 2.7 on the workers), which might lead to issues
+#     when using certain Python serializers that are incompatible across releases (e.g. marshal).
+#
+# In Spark 1.2, we removed the documentation of the IPYTHON and IPYTHON_OPTS variables, since
+# we've made the necessary changes to allow `ipython` to be used on the workers, too.  Now,
+# users can simply set PYSPARK_PYTHON=ipython to use IPython and set PYSPARK_DRIVER_PYTHON_OPTS to
+# pass options when starting the Python driver (e.g. PYSPARK_DRIVER_PYTHON_OPTS='notebook').
+#
+# For backwards-compatibility, we retain the old IPYTHON and IPYTHON_OPTS variables.
+
+# If IPython options are specified, assume user wants to run IPython (for backwards-compatibility)
+if [[ -n "$IPYTHON_OPTS" ]]; then
+  IPYTHON=1
+  # For backwards-compatibility:
+  PYSPARK_DRIVER_PYTHON_OPTS="$PYSPARK_DRIVER_PYTHON_OPTS $IPYTHON_OPTS"
 fi
-export PYSPARK_PYTHON
 
-if [[ -z "$PYSPARK_PYTHON_OPTS" && -n "$IPYTHON_OPTS" ]]; then
-  # for backward compatibility
-  PYSPARK_PYTHON_OPTS="$IPYTHON_OPTS"
+# Figure out which Python executable to use.
+# If we're not running in the legacy IPYTHON mode, then use a default PYSPARK_PYTHON
+if [[ "$IPYTHON" != "1" && -z "$PYSPARK_PYTHON" ]]; then
+  PYSPARK_PYTHON="python"
 fi
+export PYSPARK_PYTHON
 
 # Add the PySpark classes to the Python path:
 export PYTHONPATH="$SPARK_HOME/python/:$PYTHONPATH"
@@ -111,5 +126,9 @@ if [[ "$1" =~ \.py$ ]]; then
 else
   # PySpark shell requires special handling downstream
   export PYSPARK_SHELL=1
-  exec "$PYSPARK_PYTHON" $PYSPARK_PYTHON_OPTS
+  if [[ "$IPYTHON" = "1" ]]; then
+    exec "${PYSPARK_PYTHON:-ipython}" $PYSPARK_DRIVER_PYTHON_OPTS
+  else
+    exec "$PYSPARK_PYTHON" $PYSPARK_DRIVER_PYTHON_OPTS
+  fi
 fi

core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala

@@ -108,10 +108,12 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
       serverSocket = new ServerSocket(0, 1, InetAddress.getByAddress(Array(127, 0, 0, 1)))
 
       // Create and start the worker
-      val pb = new ProcessBuilder(Seq(pythonExec, "-u", "-m", "pyspark.worker"))
+      val pb = new ProcessBuilder(Seq(pythonExec, "-m", "pyspark.worker"))
       val workerEnv = pb.environment()
       workerEnv.putAll(envVars)
       workerEnv.put("PYTHONPATH", pythonPath)
+      // This is equivalent to setting the -u flag; we use it because ipython doesn't support -u:
+      workerEnv.put("PYTHONUNBUFFERED", "YES")
       val worker = pb.start()
 
       // Redirect worker stdout and stderr
@@ -149,10 +151,12 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
 
       try {
         // Create and start the daemon
-        val pb = new ProcessBuilder(Seq(pythonExec, "-u", "-m", "pyspark.daemon"))
+        val pb = new ProcessBuilder(Seq(pythonExec, "-m", "pyspark.daemon"))
         val workerEnv = pb.environment()
         workerEnv.putAll(envVars)
         workerEnv.put("PYTHONPATH", pythonPath)
+        // This is equivalent to setting the -u flag; we use it because ipython doesn't support -u:
+        workerEnv.put("PYTHONUNBUFFERED", "YES")
         daemon = pb.start()
 
         val in = new DataInputStream(daemon.getInputStream)

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

@@ -57,6 +57,7 @@ object PythonRunner {
     val builder = new ProcessBuilder(Seq(pythonExec, formattedPythonFile) ++ otherArgs)
     val env = builder.environment()
     env.put("PYTHONPATH", pythonPath)
+    // This is equivalent to setting the -u flag; we use it because ipython doesn't support -u:
     env.put("PYTHONUNBUFFERED", "YES") // value is needed to be set to a non-empty string
     env.put("PYSPARK_GATEWAY_PORT", "" + gatewayServer.getListeningPort)
     builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize

docs/programming-guide.md

@@ -217,11 +217,11 @@ use IPython, set the `PYSPARK_PYTHON` variable to `ipython` when running `bin/py
 $ PYSPARK_PYTHON=ipython ./bin/pyspark
 {% endhighlight %}
 
-You can customize the `ipython` command by setting `PYSPARK_PYTHON_OPTS`. For example, to launch
+You can customize the `ipython` command by setting `PYSPARK_DRIVER_PYTHON_OPTS`. For example, to launch
 the [IPython Notebook](http://ipython.org/notebook.html) with PyLab plot support:
 
 {% highlight bash %}
-$ PYSPARK_PYTHON=ipython PYSPARK_PYTHON_OPTS="notebook --pylab inline" ./bin/pyspark
+$ PYSPARK_PYTHON=ipython PYSPARK_DRIVER_PYTHON_OPTS="notebook --pylab inline" ./bin/pyspark
 {% endhighlight %}
 
 </div>