holdenk commented on code in PR #59778:
URL: https://github.com/apache/spark/pull/59778#discussion_r4429602721
##########
core/src/main/scala/org/apache/spark/api/python/PythonUtils.scala:
##########
@@ -196,4 +197,70 @@ private[spark] object PythonUtils extends Logging {
broadcastVars = List.empty.asJava,
accumulator = null)
}
+
+ /**
+ * Checks whether `pythonExec` can run Arrow-optimized Python UDFs, that is,
whether it has
+ * PyArrow at the version PySpark requires, and pandas too if
`requirePandas`, which the legacy
+ * pandas conversion needs. A Python client runs a similar check before it
creates such a UDF,
+ * see `_create_py_udf` in `pyspark.sql.udf`, where it always requires
pandas.
+ *
+ * It runs `pythonExec` with the environment, Python path and working
directory that
+ * `PythonWorkerFactory` gives a Python worker launched with `envVars`.
+ *
+ * @return `None` if it can, and why it cannot otherwise.
+ * @throws SparkException if the check fails otherwise, or does not finish
within `timeoutMs`.
+ */
+ private[spark] def checkArrowOptimization(
+ pythonExec: String,
+ envVars: Map[String, String],
+ requirePandas: Boolean = false,
+ timeoutMs: Long = PythonWorkerFactory.PROCESS_WAIT_TIMEOUT_MS):
Option[String] = {
+ // Exits with 3 only when PyArrow, or pandas when required, is missing or
too old, so that no
+ // other failure, such as PySpark not being importable, is taken for that.
+ val checks = (if (requirePandas) Seq("require_minimum_pandas_version")
else Nil) :+
+ "require_minimum_pyarrow_version"
+ val code =
+ s"""
+ |import sys
+ |from pyspark.sql.pandas.utils import ${checks.mkString(", ")}
+ |try:
+ | ${checks.map(_ + "()").mkString("; ")}
+ |except ImportError as e:
+ | print(e)
+ | sys.exit(3)
+ |""".stripMargin
Review Comment:
I don't love the inline Python here personally a helper python file is sort
of the vibe I'd go with but nit again
##########
core/src/main/scala/org/apache/spark/api/python/PythonUtils.scala:
##########
@@ -196,4 +197,70 @@ private[spark] object PythonUtils extends Logging {
broadcastVars = List.empty.asJava,
accumulator = null)
}
+
+ /**
+ * Checks whether `pythonExec` can run Arrow-optimized Python UDFs, that is,
whether it has
+ * PyArrow at the version PySpark requires, and pandas too if
`requirePandas`, which the legacy
+ * pandas conversion needs. A Python client runs a similar check before it
creates such a UDF,
+ * see `_create_py_udf` in `pyspark.sql.udf`, where it always requires
pandas.
+ *
+ * It runs `pythonExec` with the environment, Python path and working
directory that
+ * `PythonWorkerFactory` gives a Python worker launched with `envVars`.
+ *
+ * @return `None` if it can, and why it cannot otherwise.
+ * @throws SparkException if the check fails otherwise, or does not finish
within `timeoutMs`.
+ */
+ private[spark] def checkArrowOptimization(
+ pythonExec: String,
+ envVars: Map[String, String],
+ requirePandas: Boolean = false,
+ timeoutMs: Long = PythonWorkerFactory.PROCESS_WAIT_TIMEOUT_MS):
Option[String] = {
Review Comment:
nit: None for if it can is a bit of a funky return type and it's a
combination None or exception I'd just go bool + exception personally and make
sure we only call this once.
##########
python/docs/source/migration_guide/pyspark_upgrade.rst:
##########
@@ -19,6 +19,10 @@
Upgrading PySpark
==================
+Upgrading from PySpark 4.3 to 4.4
+---------------------------------
+* In Spark 4.4, Spark Connect falls back to a regular Python UDF when the
server's Python does not have PyArrow installed, instead of failing on an
Arrow-optimized Python UDF (the default since Spark 4.2), and logs a warning on
the server. With
``spark.sql.legacy.execution.pythonUDF.pandas.conversion.enabled`` enabled, it
also falls back when the server's Python does not have pandas installed. Like
Spark Classic, it checks the Python on the driver, so if only the executors
have these packages installed, Arrow-optimized Python UDFs now run as regular
Python UDFs. To keep them Arrow-optimized, install the packages on the driver
as well.
+
Review Comment:
I don't think we need a migration note
##########
core/src/main/scala/org/apache/spark/api/python/PythonUtils.scala:
##########
@@ -196,4 +197,70 @@ private[spark] object PythonUtils extends Logging {
broadcastVars = List.empty.asJava,
accumulator = null)
}
+
+ /**
+ * Checks whether `pythonExec` can run Arrow-optimized Python UDFs, that is,
whether it has
+ * PyArrow at the version PySpark requires, and pandas too if
`requirePandas`, which the legacy
+ * pandas conversion needs. A Python client runs a similar check before it
creates such a UDF,
+ * see `_create_py_udf` in `pyspark.sql.udf`, where it always requires
pandas.
+ *
+ * It runs `pythonExec` with the environment, Python path and working
directory that
+ * `PythonWorkerFactory` gives a Python worker launched with `envVars`.
+ *
+ * @return `None` if it can, and why it cannot otherwise.
+ * @throws SparkException if the check fails otherwise, or does not finish
within `timeoutMs`.
+ */
+ private[spark] def checkArrowOptimization(
+ pythonExec: String,
+ envVars: Map[String, String],
+ requirePandas: Boolean = false,
+ timeoutMs: Long = PythonWorkerFactory.PROCESS_WAIT_TIMEOUT_MS):
Option[String] = {
+ // Exits with 3 only when PyArrow, or pandas when required, is missing or
too old, so that no
+ // other failure, such as PySpark not being importable, is taken for that.
+ val checks = (if (requirePandas) Seq("require_minimum_pandas_version")
else Nil) :+
+ "require_minimum_pyarrow_version"
+ val code =
+ s"""
+ |import sys
+ |from pyspark.sql.pandas.utils import ${checks.mkString(", ")}
+ |try:
+ | ${checks.map(_ + "()").mkString("; ")}
+ |except ImportError as e:
+ | print(e)
+ | sys.exit(3)
+ |""".stripMargin
+ val jobArtifactUUID =
+
JobArtifactSet.getCurrentJobArtifactState.map(_.uuid).getOrElse("default")
+ val workingDir = if (jobArtifactUUID != "default") {
+ val dir = new File(SparkFiles.getRootDirectory(), jobArtifactUUID)
+ dir.mkdir()
+ dir
+ } else {
+ new File(".")
+ }
+ val pythonPath = mergePythonPaths(
+ sparkPythonPath, envVars.getOrElse("PYTHONPATH", ""),
sys.env.getOrElse("PYTHONPATH", ""))
+ val process = Utils.executeCommand(
+ Seq(pythonExec, "-c", code), workingDir, envVars + ("PYTHONPATH" ->
pythonPath))
+ try {
+ val output = new StringBuffer
+ def appendToOutput(line: String): Unit = output.append(line).append("\n")
+ val reader = Utils.processStreamByLine(
+ s"read stdout for $pythonExec", process.getInputStream, appendToOutput)
+ if (!process.waitFor(timeoutMs, TimeUnit.MILLISECONDS)) {
+ throw new SparkException(
+ s"The Arrow optimization check in $pythonExec did not finish within
$timeoutMs ms")
+ }
+ reader.join(timeoutMs)
+ process.exitValue() match {
+ case 0 => None
+ case 3 => Some(output.toString.trim)
+ case exitCode =>
+ throw new SparkException(
+ s"The Arrow optimization check in $pythonExec exited with code
$exitCode")
+ }
Review Comment:
Yeah this is weird right? either throw or don't but don't have two seperate
error paths.
##########
core/src/main/scala/org/apache/spark/api/python/PythonUtils.scala:
##########
@@ -196,4 +197,70 @@ private[spark] object PythonUtils extends Logging {
broadcastVars = List.empty.asJava,
accumulator = null)
}
+
+ /**
+ * Checks whether `pythonExec` can run Arrow-optimized Python UDFs, that is,
whether it has
+ * PyArrow at the version PySpark requires, and pandas too if
`requirePandas`, which the legacy
+ * pandas conversion needs. A Python client runs a similar check before it
creates such a UDF,
+ * see `_create_py_udf` in `pyspark.sql.udf`, where it always requires
pandas.
+ *
+ * It runs `pythonExec` with the environment, Python path and working
directory that
+ * `PythonWorkerFactory` gives a Python worker launched with `envVars`.
+ *
+ * @return `None` if it can, and why it cannot otherwise.
+ * @throws SparkException if the check fails otherwise, or does not finish
within `timeoutMs`.
+ */
+ private[spark] def checkArrowOptimization(
+ pythonExec: String,
+ envVars: Map[String, String],
+ requirePandas: Boolean = false,
+ timeoutMs: Long = PythonWorkerFactory.PROCESS_WAIT_TIMEOUT_MS):
Option[String] = {
+ // Exits with 3 only when PyArrow, or pandas when required, is missing or
too old, so that no
+ // other failure, such as PySpark not being importable, is taken for that.
+ val checks = (if (requirePandas) Seq("require_minimum_pandas_version")
else Nil) :+
+ "require_minimum_pyarrow_version"
+ val code =
+ s"""
+ |import sys
+ |from pyspark.sql.pandas.utils import ${checks.mkString(", ")}
+ |try:
+ | ${checks.map(_ + "()").mkString("; ")}
+ |except ImportError as e:
+ | print(e)
+ | sys.exit(3)
+ |""".stripMargin
+ val jobArtifactUUID =
+
JobArtifactSet.getCurrentJobArtifactState.map(_.uuid).getOrElse("default")
+ val workingDir = if (jobArtifactUUID != "default") {
+ val dir = new File(SparkFiles.getRootDirectory(), jobArtifactUUID)
+ dir.mkdir()
+ dir
+ } else {
+ new File(".")
+ }
+ val pythonPath = mergePythonPaths(
+ sparkPythonPath, envVars.getOrElse("PYTHONPATH", ""),
sys.env.getOrElse("PYTHONPATH", ""))
+ val process = Utils.executeCommand(
+ Seq(pythonExec, "-c", code), workingDir, envVars + ("PYTHONPATH" ->
pythonPath))
Review Comment:
We should have a helper to launch a Python file this will drift from the
main path
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
