You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@kyuubi.apache.org by GitBox <gi...@apache.org> on 2022/11/04 05:42:11 UTC
[GitHub] [incubator-kyuubi] cfmcgrady opened a new pull request, #3762: PySpark support
cfmcgrady opened a new pull request, #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762
<!--
Thanks for sending a pull request!
Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/CONTRIBUTING.html
2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->
### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
1. If you add a feature, you can talk about the use case of it.
2. If you fix a bug, you can clarify why it is a bug.
-->
### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [ ] Add screenshots for manual tests if appropriate
- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015352781
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ t1: Thread,
+ t2: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ t1.interrupt()
+ t2.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
+
+ def createSessionPythonWorker(): SessionPythonWorker = {
+ val builder = new ProcessBuilder(Seq(
+ pythonExec,
+ s"${ExecutePython.kyuubiPythonPath}/execute_python.py").asJava)
+ val env = builder.environment()
+ val pythonPath = sys.env.getOrElse("PYTHONPATH", "")
+ .split(File.pathSeparator)
+ .++(ExecutePython.kyuubiPythonPath.toString)
+ env.put("PYTHONPATH", pythonPath.mkString(File.pathSeparator))
+ env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", defaultSparkHome()))
+ env.put("PYTHON_GATEWAY_CONNECTION_INFO", KyuubiPythonGatewayServer.CONNECTION_FILE_PATH)
+ logger.info(
+ s"""
+ |launch python worker command: ${builder.command().asScala.mkString(" ")}
+ |environment:
+ |${builder.environment().asScala.map(kv => kv._1 + "=" + kv._2).mkString("\n")}
+ |""".stripMargin)
+ builder.redirectError(Redirect.PIPE)
+ val process = builder.start()
+ SessionPythonWorker(start_stderr_steam_reader(process), start_watcher(process), process)
+ }
+
+ // for test
+ def defaultSparkHome(): String = {
+ val homeDirFilter: FilenameFilter = (dir: File, name: String) =>
+ dir.isDirectory && name.contains("spark-") && !name.contains("-engine")
+ // get from kyuubi-server/../externals/kyuubi-download/target
+ new File(getClass.getProtectionDomain.getCodeSource.getLocation.toURI).getPath
+ .split("kyuubi-spark-sql-engine").flatMap { cwd =>
+ val candidates = Paths.get(cwd, "kyuubi-download", "target")
+ .toFile.listFiles(homeDirFilter)
+ if (candidates == null) None else candidates.map(_.toPath).headOption
+ }.find(Files.exists(_)).map(_.toAbsolutePath.toFile.getCanonicalPath)
+ .getOrElse {
+ throw new IllegalStateException("SPARK_HOME not found!")
+ }
+ }
+
+ private def start_stderr_steam_reader(process: Process): Thread = {
Review Comment:
why use such naming style?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014650567
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
+
+from py4j.java_gateway import java_import, JavaGateway, JavaObject, GatewayParameters
+from py4j.clientserver import ClientServer, JavaParameters, PythonParameters
+from pyspark.context import SparkContext
+from pyspark.serializers import read_int, write_with_length, UTF8Deserializer
+from pyspark.sql import SparkSession
+
+
+def connect_to_exist_gateway():
+ conn_info_file = '/tmp/connection.info'
Review Comment:
how about passing the path by env?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
pan3793 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1306606916
I think it's already in a good shape to get into master, thanks @cfmcgrady and all reviewers.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016156153
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
```suggestion
private val kyuubiPythonPath = Files.createTempDirectory("")
lazy val init: Unit = {
KyuubiPythonGatewayServer.start()
writeTempPyFile(kyuubiPythonPath, "execute_python.py")
writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
}
```
How about simply use `lazy val` to lock the object instance instead. @cfmcgrady @pan3793
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] dik111 commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
dik111 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304946747
> > will it suport kyuubi jdbc in the future？
>
> what do you mean `kyuubi jdbc`?
I mean can I run pyspark code through jdbc or rest api, because we use Kyuubi as an online SQL gateway
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014863236
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
I believe you copy this method from the superclass w/ a little bit of modification, but it's hard to distinguish what's exactly modified here. How about making a prepared PR to copy the original method from the superclass? This could benefit reviewers.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014968811
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/api/python/KyuubiPythonGatewayServer.scala:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.api.python
+
+import java.io.{DataOutputStream, File, FileOutputStream}
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.Files
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+
+object KyuubiPythonGatewayServer extends Logging {
+
+ val CONNECTION_FILE_PATH = "/tmp/connection.info"
Review Comment:
to avoid multiple Spark engine conflicts, create a temporary directory each time.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016156153
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
```suggestion
private val kyuubiPythonPath = Files.createTempDirectory("")
lazy val init: Unit = {
KyuubiPythonGatewayServer.start()
writeTempPyFile(kyuubiPythonPath, "execute_python.py")
writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
}
```
simply to use `lazy val` to lock the object instance instead.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016150455
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
Review Comment:
Is it thread safe here? it's locking the code block itself rather than syncing the AtomicBoolean and with no double check.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015628827
##########
externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala:
##########
@@ -683,6 +683,16 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
}
}
+ test("pyspark support") {
+ withMultipleConnectionJdbcStatement()({ statement =>
Review Comment:
added.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014650602
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
+
+from py4j.java_gateway import java_import, JavaGateway, JavaObject, GatewayParameters
+from py4j.clientserver import ClientServer, JavaParameters, PythonParameters
+from pyspark.context import SparkContext
+from pyspark.serializers import read_int, write_with_length, UTF8Deserializer
+from pyspark.sql import SparkSession
+
+
+def connect_to_exist_gateway():
+ conn_info_file = '/tmp/connection.info'
+ with open(conn_info_file, "rb") as info:
+ gateway_port = read_int(info)
+ gateway_secret = UTF8Deserializer().loads(info)
+ if os.environ.get("PYSPARK_PIN_THREAD", "true").lower() == "true":
+ gateway = ClientServer(
+ java_parameters=JavaParameters(
+ port=gateway_port,
+ auth_token=gateway_secret,
+ auto_convert=True),
+ python_parameters=PythonParameters(
+ port=0,
+ eager_load=False))
+ else:
+ gateway = JavaGateway(
+ gateway_parameters=GatewayParameters(
+ port=gateway_port,
+ auth_token=gateway_secret,
+ auto_convert=True))
+ # gateway.proc = proc
+
+ # Import the classes used by PySpark
+ java_import(gateway.jvm, "org.apache.spark.SparkConf")
+ java_import(gateway.jvm, "org.apache.spark.api.java.*")
+ java_import(gateway.jvm, "org.apache.spark.api.python.*")
+ java_import(gateway.jvm, "org.apache.spark.ml.python.*")
+ java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
+ java_import(gateway.jvm, "org.apache.spark.resource.*")
+ # TODO(davies): move into sql
Review Comment:
.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304836073
> will it suport kyuubi jdbc in the future？
what do you mean `kyuubi jdbc`?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016190827
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
> How does the `lazy val` expression lock the object instance?
FYI: https://www.baeldung.com/scala/lazy-val
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304845026
> will it suport kyuubi jdbc in the future？
From my understanding, the current implementation supports both Hive Jdbc driver and Kyuubi Jdbc driver, but things are different in the Beeline case.
Beeline is responsible to split SQL(python/scala code) before sending each query through the JDBC statement, and in default, Beeline calls `trim` for each query before sending, it may break the python code since python is indenting sensitive. That's why this PR also modified some code in the Beeline module but does not touch the JDBC module.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016150455
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
Review Comment:
Is it possible double initialization here? it's locking the code block itself rather than syncing the AtomicBoolean and with no double check.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304950777
I think it's a YES. The python script is sent on JDBC connection in screenshots, just exact same way as the SQL or scala scripts. But it seems like a drawback with a mandatory semicolon at the end of submission to inform SQL cli standard. @dik111
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] codecov-commenter commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1305760574
# [Codecov](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3762](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (1c1cf31) into [master](https://codecov.io/gh/apache/incubator-kyuubi/commit/ce74e27a506dcd48fe61bad30f98358252659c8b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ce74e27) will **increase** coverage by `0.14%`.
> The diff coverage is `75.26%`.
> :exclamation: Current head 1c1cf31 differs from pull request most recent head aa6aedf. Consider uploading reports for the commit aa6aedf to get more accurate results
```diff
@@ Coverage Diff @@
## master #3762 +/- ##
============================================
+ Coverage 52.63% 52.78% +0.14%
Complexity 13 13
============================================
Files 494 496 +2
Lines 27800 27976 +176
Branches 3835 3859 +24
============================================
+ Hits 14633 14767 +134
- Misses 11774 11810 +36
- Partials 1393 1399 +6
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...n/java/org/apache/hive/beeline/KyuubiCommands.java](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWhpdmUtYmVlbGluZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaGl2ZS9iZWVsaW5lL0t5dXViaUNvbW1hbmRzLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...e/spark/api/python/KyuubiPythonGatewayServer.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvc3BhcmsvYXBpL3B5dGhvbi9LeXV1YmlQeXRob25HYXRld2F5U2VydmVyLnNjYWxh) | `73.91% <73.91%> (ø)` | |
| [...ine/spark/operation/SparkSQLOperationManager.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9vcGVyYXRpb24vU3BhcmtTUUxPcGVyYXRpb25NYW5hZ2VyLnNjYWxh) | `85.71% <90.00%> (+0.63%)` | :arrow_up: |
| [.../kyuubi/engine/spark/operation/ExecutePython.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9vcGVyYXRpb24vRXhlY3V0ZVB5dGhvbi5zY2FsYQ==) | `90.90% <90.90%> (ø)` | |
| [...kyuubi/engine/spark/session/SparkSessionImpl.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXh0ZXJuYWxzL2t5dXViaS1zcGFyay1zcWwtZW5naW5lL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUva3l1dWJpL2VuZ2luZS9zcGFyay9zZXNzaW9uL1NwYXJrU2Vzc2lvbkltcGwuc2NhbGE=) | `63.41% <100.00%> (+1.87%)` | :arrow_up: |
| [...in/scala/org/apache/kyuubi/config/KyuubiConf.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWNvbW1vbi9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9jb25maWcvS3l1dWJpQ29uZi5zY2FsYQ==) | `97.46% <100.00%> (-0.08%)` | :arrow_down: |
| [...ache/kyuubi/operation/KyuubiOperationManager.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9vcGVyYXRpb24vS3l1dWJpT3BlcmF0aW9uTWFuYWdlci5zY2FsYQ==) | `79.45% <0.00%> (-2.74%)` | :arrow_down: |
| [...ache/kyuubi/server/KyuubiRestFrontendService.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXJ2ZXIvS3l1dWJpUmVzdEZyb250ZW5kU2VydmljZS5zY2FsYQ==) | `81.39% <0.00%> (-2.52%)` | :arrow_down: |
| [...rg/apache/kyuubi/ctl/cmd/log/LogBatchCommand.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLWN0bC9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9jdGwvY21kL2xvZy9Mb2dCYXRjaENvbW1hbmQuc2NhbGE=) | `63.93% <0.00%> (-1.64%)` | :arrow_down: |
| [.../org/apache/kyuubi/session/KyuubiSessionImpl.scala](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-a3l1dWJpLXNlcnZlci9zcmMvbWFpbi9zY2FsYS9vcmcvYXBhY2hlL2t5dXViaS9zZXNzaW9uL0t5dXViaVNlc3Npb25JbXBsLnNjYWxh) | `75.22% <0.00%> (-0.89%)` | :arrow_down: |
| ... and [15 more](https://codecov.io/gh/apache/incubator-kyuubi/pull/3762/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014650484
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
Review Comment:
where is `xrange` used?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014647714
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/api/python/KyuubiPythonGatewayServer.scala:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.api.python
+
+import java.io.{DataOutputStream, File, FileOutputStream}
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.Files
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+
+object KyuubiPythonGatewayServer extends Logging {
+
+ val CONNECTION_FILE_PATH = "/tmp/connection.info"
Review Comment:
```suggestion
val CONNECTION_FILE_PATH = s"${sys.props("java.io.tmpdir")}/connection.info"
```
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014974081
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
reverted changes by https://github.com/apache/incubator-kyuubi/pull/3762/commits/1c1cf31d2b5e296ef4df7307433880f484800c8a and filed a new issue #3770 for tracking this
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014863236
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
I believe you copy this method from the superclass w/ a little bit of modification, but it's hard to distinguish what's exactly modified here. How about making a paving PR to copy the original method from the superclass? This could benefit reviewers.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016178567
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
Review Comment:
added double-check.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016228289
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
But more readness burden for one-time usage, and introduced more memory footprint for this single check.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016179717
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
How does the `lazy val` expression lock the object instance?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016188113
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
`lazy val` in Scala basically equivalent to auto added method in compile time as `synchronized (this) + double check empty` in Java
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016192236
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
> How about simply use lazy val to lock the object instance instead.
I think `double-check` style is more readable than `lazy val`, WDYT? cc @pan3793
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 closed pull request #3762: [KYUUBI #3782][PYSPARK] Initial support PySpark
Posted by GitBox <gi...@apache.org>.
pan3793 closed pull request #3762: [KYUUBI #3782][PYSPARK] Initial support PySpark
URL: https://github.com/apache/incubator-kyuubi/pull/3762
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016099095
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
it's supported now
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015351190
##########
externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala:
##########
@@ -683,6 +683,16 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
}
}
+ test("pyspark support") {
+ withMultipleConnectionJdbcStatement()({ statement =>
Review Comment:
we should assume `python3` is available before running this test
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] turboFei commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
turboFei commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304468120
fixed the ci issue, could you rebase this pr on master?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014650094
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ t1: Thread,
+ t2: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ t1.interrupt()
+ t2.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+ // val pythonExec = "/usr/local/share/anaconda3/envs/pyspark-3.3.1-py36/bin/python"
+
+ // TODO:(fchen) get from conf
+ val pythonExec = sys.env.get("PYSPARK_PYTHON")
Review Comment:
```suggestion
val pythonExec = sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
```
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1303030641
This pyspark support capacity is what we desired.
Let's follow this PR to see whether it suits our situation.
cc @jeanlyn
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: [KYUUBI #3782][PYSPARK] Initial support PySpark
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016537547
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ errorReader: Thread,
+ pythonWorkerMonitor: Thread,
+ workerProcess: Process) {
+ private val stdin: PrintWriter = new PrintWriter(workerProcess.getOutputStream)
+ private val stdout: BufferedReader =
+ new BufferedReader(new InputStreamReader(workerProcess.getInputStream), 1)
+
+ def runCode(code: String): Option[PythonReponse] = {
+ val input = ExecutePython.toJson(Map("code" -> code, "cmd" -> "run_code"))
+ // scalastyle:off println
+ stdin.println(input)
+ // scalastyle:on
+ stdin.flush()
+ Option(stdout.readLine())
+ .map(ExecutePython.fromJson[PythonReponse](_))
+ }
+
+ def close(): Unit = {
+ val exitCmd = ExecutePython.toJson(Map("cmd" -> "exit_worker"))
+ // scalastyle:off println
+ stdin.println(exitCmd)
+ // scalastyle:on
+ stdin.flush()
+ stdin.close()
+ stdout.close()
+ errorReader.interrupt()
+ pythonWorkerMonitor.interrupt()
+ workerProcess.destroy()
+ }
+}
+
+object ExecutePython extends Logging {
+
+ // TODO:(fchen) get from conf
+ val pythonExec =
+ sys.env.getOrElse("PYSPARK_PYTHON", sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", "python3"))
+ private val isPythonGatewayStart = new AtomicBoolean(false)
+ val kyuubiPythonPath = Files.createTempDirectory("")
+ def init(): Unit = {
+ if (!isPythonGatewayStart.get()) {
+ synchronized {
+ KyuubiPythonGatewayServer.start()
+ writeTempPyFile(kyuubiPythonPath, "execute_python.py")
+ writeTempPyFile(kyuubiPythonPath, "kyuubi_util.py")
+ isPythonGatewayStart.set(true)
+ }
+ }
+ }
Review Comment:
either is fine to me.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015354198
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecutePython.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.
+ */
+
+package org.apache.kyuubi.engine.spark.operation
+
+import java.io.{BufferedReader, File, FilenameFilter, FileOutputStream, InputStreamReader, PrintWriter}
+import java.lang.ProcessBuilder.Redirect
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.spark.api.python.KyuubiPythonGatewayServer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.StructType
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.operation.ArrayFetchIterator
+import org.apache.kyuubi.session.Session
+
+class ExecutePython(
+ session: Session,
+ override val statement: String,
+ worker: SessionPythonWorker) extends SparkOperation(session) {
+
+ override protected def resultSchema: StructType = {
+ if (result == null || result.schema.isEmpty) {
+ new StructType().add("output", "string")
+ .add("status", "string")
+ .add("ename", "string")
+ .add("evalue", "string")
+ .add("traceback", "array<string>")
+ } else {
+ result.schema
+ }
+ }
+
+ override protected def runInternal(): Unit = {
+ val response = worker.runCode(statement)
+ val output = response.map(_.content.getOutput()).getOrElse("")
+ val status = response.map(_.content.status).getOrElse("UNKNOWN_STATUS")
+ val ename = response.map(_.content.getEname()).getOrElse("")
+ val evalue = response.map(_.content.getEvalue()).getOrElse("")
+ val traceback = response.map(_.content.getTraceback()).getOrElse(Array.empty)
+ iter =
+ new ArrayFetchIterator[Row](Array(Row(output, status, ename, evalue, Row(traceback: _*))))
+ }
+
+}
+
+case class SessionPythonWorker(
+ t1: Thread,
+ t2: Thread,
Review Comment:
could you give t1 t2 a meaningful name?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015357567
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
Review Comment:
it calls `range` instead of `xrange`, so I think it's redundant. And we are not tend to support Python2, all `if sys.version >= '3'` could be removed.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] dik111 commented on pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
dik111 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1304746107
will it suport kyuubi jdbc in the future？
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015546407
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
Review Comment:
updated.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] pan3793 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1015355352
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,88 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
+
+from py4j.java_gateway import java_import, JavaGateway, JavaObject, GatewayParameters
+from py4j.clientserver import ClientServer, JavaParameters, PythonParameters
+from pyspark.context import SparkContext
+from pyspark.serializers import read_int, write_with_length, UTF8Deserializer
+from pyspark.sql import SparkSession
+
+
+def connect_to_exist_gateway():
+ conn_info_file = os.environ.get("PYTHON_GATEWAY_CONNECTION_INFO", "/tmp/connection.info")
Review Comment:
throw exception to fail fast
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014968251
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
+
+from py4j.java_gateway import java_import, JavaGateway, JavaObject, GatewayParameters
+from py4j.clientserver import ClientServer, JavaParameters, PythonParameters
+from pyspark.context import SparkContext
+from pyspark.serializers import read_int, write_with_length, UTF8Deserializer
+from pyspark.sql import SparkSession
+
+
+def connect_to_exist_gateway():
+ conn_info_file = '/tmp/connection.info'
Review Comment:
updated.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1014964993
##########
externals/kyuubi-spark-sql-engine/src/main/resources/python/kyuubi_util.py:
##########
@@ -0,0 +1,89 @@
+#
+# 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.
+#
+
+import atexit
+import os
+import sys
+import signal
+import shlex
+import shutil
+import socket
+import platform
+import tempfile
+import time
+from subprocess import Popen, PIPE
+
+if sys.version >= '3':
+ xrange = range
Review Comment:
It is used in L137 by function `execute_reply_error(exc_type, exc_value, tb)`
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016098546
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
Shall we add unspported multi-line scripts to limitation in this PR description?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] cfmcgrady commented on a diff in pull request #3762: PySpark support
Posted by GitBox <gi...@apache.org>.
cfmcgrady commented on code in PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#discussion_r1016097130
##########
kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiCommands.java:
##########
@@ -560,4 +558,56 @@ public void run() {
}
}
}
+
+ @Override
+ public String handleMultiLineCmd(String line) throws IOException {
Review Comment:
addressed in https://github.com/apache/incubator-kyuubi/pull/3762/commits/3e4d6e3fc05c641e519ac41d2330bc05dfb61d7f
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] bowenliang123 commented on pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
bowenliang123 commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1306615716
LGTM. And locally tested.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
[GitHub] [incubator-kyuubi] yaooqinn commented on pull request #3762: Initial support PySpark
Posted by GitBox <gi...@apache.org>.
yaooqinn commented on PR #3762:
URL: https://github.com/apache/incubator-kyuubi/pull/3762#issuecomment-1306742291
Let's create a ticket to track for such a big feature.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org