diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md
index b42bdcf10..07ae62771 100644
--- a/docs/configuration/settings.md
+++ b/docs/configuration/settings.md
@@ -181,6 +181,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
| kyuubi.engine.single.spark.session | false | When set to true, this engine is running in a single session mode. All the JDBC/ODBC connections share the temporary views, function registries, SQL configuration and the current database. | boolean | 1.3.0 |
| kyuubi.engine.spark.event.loggers | SPARK | A comma-separated list of engine loggers, where engine/session/operation etc events go. - SPARK: the events will be written to the Spark listener bus.
- JSON: the events will be written to the location of kyuubi.engine.event.json.log.path
- JDBC: to be done
- CUSTOM: to be done.
| seq | 1.7.0 |
| kyuubi.engine.spark.initialize.sql | SHOW DATABASES | The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`. | seq | 1.8.1 |
+| kyuubi.engine.spark.output.mode | AUTO | The output mode of Spark engine: - AUTO: For PySpark, the extracted `text/plain` from python response as output.
- NOTEBOOK: For PySpark, the original python response as output.
| string | 1.9.0 |
| kyuubi.engine.spark.python.env.archive | <undefined> | Portable Python env archive used for Spark engine Python language mode. | string | 1.7.0 |
| kyuubi.engine.spark.python.env.archive.exec.path | bin/python | The Python exec path under the Python env archive. | string | 1.7.0 |
| kyuubi.engine.spark.python.home.archive | <undefined> | Spark archive containing $SPARK_HOME/python directory, which is used to init session Python worker for Python language mode. | string | 1.7.0 |
diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala
index 7009bf9e0..f60b1d4c8 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.types.StructType
import org.apache.kyuubi.{KyuubiSQLException, Logging, Utils}
import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SPARK_PYTHON_ENV_ARCHIVE, ENGINE_SPARK_PYTHON_ENV_ARCHIVE_EXEC_PATH, ENGINE_SPARK_PYTHON_HOME_ARCHIVE, ENGINE_SPARK_PYTHON_MAGIC_ENABLED}
+import org.apache.kyuubi.config.KyuubiConf.EngineSparkOutputMode.{AUTO, EngineSparkOutputMode, NOTEBOOK}
import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_SESSION_USER_KEY, KYUUBI_STATEMENT_ID_KEY}
import org.apache.kyuubi.engine.spark.KyuubiSparkUtil._
import org.apache.kyuubi.engine.spark.util.JsonUtils
@@ -86,7 +87,7 @@ class ExecutePython(
val response = worker.runCode(statement)
val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
if (PythonResponse.OK_STATUS.equalsIgnoreCase(status)) {
- val output = response.map(_.content.getOutput()).getOrElse("")
+ val output = response.map(_.content.getOutput(outputMode)).getOrElse("")
val ename = response.map(_.content.getEname()).getOrElse("")
val evalue = response.map(_.content.getEvalue()).getOrElse("")
val traceback = response.map(_.content.getTraceback()).getOrElse(Seq.empty)
@@ -403,18 +404,22 @@ case class PythonResponseContent(
evalue: String,
traceback: Seq[String],
status: String) {
- def getOutput(): String = {
+ def getOutput(outputMode: EngineSparkOutputMode): String = {
if (data == null) return ""
- // If data does not contains field other than `test/plain`, keep backward compatibility,
- // otherwise, return all the data.
- if (data.filterNot(_._1 == "text/plain").isEmpty) {
- data.get("text/plain").map {
- case str: String => str
- case obj => JsonUtils.toJson(obj)
- }.getOrElse("")
- } else {
- JsonUtils.toJson(data)
+ outputMode match {
+ case AUTO =>
+ // If data does not contains field other than `test/plain`, keep backward compatibility,
+ // otherwise, return all the data.
+ if (data.filterNot(_._1 == "text/plain").isEmpty) {
+ data.get("text/plain").map {
+ case str: String => str
+ case obj => JsonUtils.toJson(obj)
+ }.getOrElse("")
+ } else {
+ JsonUtils.toJson(data)
+ }
+ case NOTEBOOK => JsonUtils.toJson(data)
}
}
def getEname(): String = {
diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkOperation.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkOperation.scala
index 1d271cfce..88ebc306b 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkOperation.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkOperation.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.types.{BinaryType, StructField, StructType}
import org.apache.kyuubi.{KyuubiSQLException, Utils}
import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.config.KyuubiConf.{ARROW_BASED_ROWSET_TIMESTAMP_AS_STRING, OPERATION_SPARK_LISTENER_ENABLED, SESSION_PROGRESS_ENABLE, SESSION_USER_SIGN_ENABLED}
+import org.apache.kyuubi.config.KyuubiConf.{ARROW_BASED_ROWSET_TIMESTAMP_AS_STRING, ENGINE_SPARK_OUTPUT_MODE, EngineSparkOutputMode, OPERATION_SPARK_LISTENER_ENABLED, SESSION_PROGRESS_ENABLE, SESSION_USER_SIGN_ENABLED}
import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_SESSION_SIGN_PUBLICKEY, KYUUBI_SESSION_USER_KEY, KYUUBI_SESSION_USER_SIGN, KYUUBI_STATEMENT_ID_KEY}
import org.apache.kyuubi.engine.spark.KyuubiSparkUtil.{getSessionConf, SPARK_SCHEDULER_POOL_KEY}
import org.apache.kyuubi.engine.spark.events.SparkOperationEvent
@@ -82,6 +82,9 @@ abstract class SparkOperation(session: Session)
protected def supportProgress: Boolean = false
+ protected def outputMode: EngineSparkOutputMode.EngineSparkOutputMode =
+ EngineSparkOutputMode.withName(getSessionConf(ENGINE_SPARK_OUTPUT_MODE, spark))
+
override def getStatus: OperationStatus = {
if (progressEnable && supportProgress) {
val progressMonitor = new SparkProgressMonitor(spark, statementId)
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
index fd01e718c..f9bca31ce 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
@@ -3258,6 +3258,20 @@ object KyuubiConf {
.booleanConf
.createWithDefault(true)
+ object EngineSparkOutputMode extends Enumeration {
+ type EngineSparkOutputMode = Value
+ val AUTO, NOTEBOOK = Value
+ }
+
+ val ENGINE_SPARK_OUTPUT_MODE: ConfigEntry[String] =
+ buildConf("kyuubi.engine.spark.output.mode")
+ .doc("The output mode of Spark engine: " +
+ " - AUTO: For PySpark, the extracted `text/plain` from python response as output.
" +
+ " - NOTEBOOK: For PySpark, the original python response as output.
")
+ .version("1.9.0")
+ .stringConf
+ .createWithDefault(EngineSparkOutputMode.AUTO.toString)
+
val ENGINE_SPARK_REGISTER_ATTRIBUTES: ConfigEntry[Seq[String]] =
buildConf("kyuubi.engine.spark.register.attributes")
.internal