From da2c07af33b1c4b38c1cc39668b662a506f4e373 Mon Sep 17 00:00:00 2001 From: "qiaoshi.wang" Date: Mon, 8 Oct 2018 15:34:35 +0800 Subject: [PATCH 1/5] add pyspark Completion function --- repl/src/main/resources/fake_shell.py | 20 +++++++++++++++++++ .../livy/repl/PySparkJobProcessor.scala | 2 ++ .../apache/livy/repl/PythonInterpreter.scala | 5 +++++ 3 files changed, 27 insertions(+) diff --git a/repl/src/main/resources/fake_shell.py b/repl/src/main/resources/fake_shell.py index 2a99185b1..8f463eef3 100644 --- a/repl/src/main/resources/fake_shell.py +++ b/repl/src/main/resources/fake_shell.py @@ -46,9 +46,21 @@ global_dict = {} job_context = None local_tmp_dir_path = None +completer = None TOP_FRAME_REGEX = re.compile(r'\s*File "".*in ') +def get_completer(): + try: + __IPYTHON__ + from IPython.core.completer import IPCompleter + ip = get_ipython() + completer = IPCompleter(ip,global_namespace=global_dict) + except NameError: + ##todo complete function can be runed in python + pass + + def execute_reply(status, content): return { 'msg_type': 'execute_reply', @@ -195,6 +207,14 @@ def processBypassJob(self, serialized_job): def addFile(self, uri_path): job_context.sc.addFile(uri_path) + + + + def complete(self,code,cursor_pos=None): + result = completer.complete(code,cursor_pos=cursor_pos)[1] + return ",".join(result) + + def addPyFile(self, uri_path): job_context.sc.addPyFile(uri_path) diff --git a/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala b/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala index 95b91cffb..f24a64d10 100644 --- a/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala +++ b/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala @@ -24,4 +24,6 @@ trait PySparkJobProcessor { def addPyFile(path: String) def getLocalTmpDirPath: String + + def complete(code:String,cursor_pos:Int):String } diff --git a/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala b/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala index bab2bfa7b..3d150d811 100644 --- a/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala +++ b/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala @@ -299,6 +299,11 @@ private class PythonInterpreter( } } + override def complete(code: String, cursor: Int): Array[String] ={ + val r = pysparkJobProcessor.complete(code,cursor) + r.split(",") + } + private def updatePythonGatewayPort(port: Int): Unit = { // The python gateway port can be 0 only when LivyConf.TEST_MODE is true // Py4j 0.10 has different API signature for "getCallbackClient", use reflection to handle it. From 039a63ad05780a7ae3421c25f0d97dbdd1db23f1 Mon Sep 17 00:00:00 2001 From: "qiaoshi.wang" Date: Mon, 8 Oct 2018 16:15:29 +0800 Subject: [PATCH 2/5] mkdify scala style --- .../main/scala/org/apache/livy/repl/PySparkJobProcessor.scala | 2 +- .../main/scala/org/apache/livy/repl/PythonInterpreter.scala | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala b/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala index f24a64d10..1154cc108 100644 --- a/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala +++ b/repl/src/main/scala/org/apache/livy/repl/PySparkJobProcessor.scala @@ -25,5 +25,5 @@ trait PySparkJobProcessor { def getLocalTmpDirPath: String - def complete(code:String,cursor_pos:Int):String + def complete(code: String, cursor_pos: Int): String } diff --git a/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala b/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala index 3d150d811..75d8593b7 100644 --- a/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala +++ b/repl/src/main/scala/org/apache/livy/repl/PythonInterpreter.scala @@ -299,8 +299,8 @@ private class PythonInterpreter( } } - override def complete(code: String, cursor: Int): Array[String] ={ - val r = pysparkJobProcessor.complete(code,cursor) + override def complete(code: String, cursor: Int): Array[String] = { + val r = pysparkJobProcessor.complete(code, cursor) r.split(",") } From e4ced915afde8451bcced451f47a5e452e285480 Mon Sep 17 00:00:00 2001 From: "qiaoshi.wang" Date: Mon, 15 Oct 2018 09:35:29 +0800 Subject: [PATCH 3/5] =?UTF-8?q?=E5=A2=9E=E5=8A=A0python=E7=89=88=E6=9C=AC?= =?UTF-8?q?=E7=9A=84=E4=BB=A3=E7=A0=81=E6=8F=90=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- repl/src/main/resources/fake_shell.py | 31 +++++++++++++++++++++------ 1 file changed, 25 insertions(+), 6 deletions(-) diff --git a/repl/src/main/resources/fake_shell.py b/repl/src/main/resources/fake_shell.py index 8f463eef3..2985b16f0 100644 --- a/repl/src/main/resources/fake_shell.py +++ b/repl/src/main/resources/fake_shell.py @@ -46,7 +46,6 @@ global_dict = {} job_context = None local_tmp_dir_path = None -completer = None TOP_FRAME_REGEX = re.compile(r'\s*File "".*in ') @@ -55,10 +54,15 @@ def get_completer(): __IPYTHON__ from IPython.core.completer import IPCompleter ip = get_ipython() - completer = IPCompleter(ip,global_namespace=global_dict) + ip_completer = IPCompleter(ip,global_namespace=global_dict) + global_dict['ip_completer'] = ip_completer except NameError: - ##todo complete function can be runed in python - pass + try: + from rlcompleter import Completer + p_completer = Completer(namespace=global_dict) + global_dict['p_completer'] = p_completer + except: + pass def execute_reply(status, content): @@ -211,8 +215,22 @@ def addFile(self, uri_path): def complete(self,code,cursor_pos=None): - result = completer.complete(code,cursor_pos=cursor_pos)[1] - return ",".join(result) + results = [] + ip_completer =global_dict.get('ip_completer') + p_completer = global_dict.get("p_completer") + if ip_completer is not None: + results = ip_completer.complete(code,cursor_pos=cursor_pos)[1] + elif p_completer is not None: + p_completer = global_dict.get("p_completer") + state = 0 + result = None + while state == 0 or result is not None: + result = p_completer.complete(code,state) + if result is not None: + results.append(result) + state += 1 + return ",".join(results) + def addPyFile(self, uri_path): @@ -575,6 +593,7 @@ def main(): sys.stderr = UnicodeDecodingStringIO() spark_major_version = os.getenv("LIVY_SPARK_MAJOR_VERSION") + get_completer() try: listening_port = 0 if os.environ.get("LIVY_TEST") != "true": From 44570f5fcbc765b3aab47e7930a17f28613dd811 Mon Sep 17 00:00:00 2001 From: "qiaoshi.wang" Date: Sun, 20 Jan 2019 22:46:36 +0800 Subject: [PATCH 4/5] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=A0=B7=E5=BC=8F?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- repl/src/main/resources/fake_shell.py | 20 ++++++++------------ 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/repl/src/main/resources/fake_shell.py b/repl/src/main/resources/fake_shell.py index 2985b16f0..539c4a241 100644 --- a/repl/src/main/resources/fake_shell.py +++ b/repl/src/main/resources/fake_shell.py @@ -49,12 +49,13 @@ TOP_FRAME_REGEX = re.compile(r'\s*File "".*in ') -def get_completer(): + +def initialize_completer(): try: __IPYTHON__ from IPython.core.completer import IPCompleter ip = get_ipython() - ip_completer = IPCompleter(ip,global_namespace=global_dict) + ip_completer = IPCompleter(ip, global_namespace=global_dict) global_dict['ip_completer'] = ip_completer except NameError: try: @@ -211,28 +212,23 @@ def processBypassJob(self, serialized_job): def addFile(self, uri_path): job_context.sc.addFile(uri_path) - - - - def complete(self,code,cursor_pos=None): + def complete(self, code, cursor_pos=None): results = [] - ip_completer =global_dict.get('ip_completer') + ip_completer = global_dict.get('ip_completer') p_completer = global_dict.get("p_completer") if ip_completer is not None: - results = ip_completer.complete(code,cursor_pos=cursor_pos)[1] + results = ip_completer.complete(code, cursor_pos=cursor_pos)[1] elif p_completer is not None: p_completer = global_dict.get("p_completer") state = 0 result = None while state == 0 or result is not None: - result = p_completer.complete(code,state) + result = p_completer.complete(code, state) if result is not None: results.append(result) state += 1 return ",".join(results) - - def addPyFile(self, uri_path): job_context.sc.addPyFile(uri_path) @@ -593,7 +589,7 @@ def main(): sys.stderr = UnicodeDecodingStringIO() spark_major_version = os.getenv("LIVY_SPARK_MAJOR_VERSION") - get_completer() + initialize_completer() try: listening_port = 0 if os.environ.get("LIVY_TEST") != "true": From 1cd30e1ff29669c1d65b7c1b18f036ddbf09354c Mon Sep 17 00:00:00 2001 From: "qiaoshi.wang" Date: Sat, 2 Feb 2019 11:21:49 +0800 Subject: [PATCH 5/5] =?UTF-8?q?=E6=B7=BB=E5=8A=A0ip=5Fcompleter,=20p=5Fcom?= =?UTF-8?q?pleter=E5=8F=98=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- repl/src/main/resources/fake_shell.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/repl/src/main/resources/fake_shell.py b/repl/src/main/resources/fake_shell.py index 539c4a241..cd7129264 100644 --- a/repl/src/main/resources/fake_shell.py +++ b/repl/src/main/resources/fake_shell.py @@ -46,22 +46,23 @@ global_dict = {} job_context = None local_tmp_dir_path = None +ip_completer = None +p_completer = None TOP_FRAME_REGEX = re.compile(r'\s*File "".*in ') def initialize_completer(): + global ip_completer,p_completer try: __IPYTHON__ from IPython.core.completer import IPCompleter ip = get_ipython() ip_completer = IPCompleter(ip, global_namespace=global_dict) - global_dict['ip_completer'] = ip_completer except NameError: try: from rlcompleter import Completer p_completer = Completer(namespace=global_dict) - global_dict['p_completer'] = p_completer except: pass @@ -214,12 +215,9 @@ def addFile(self, uri_path): def complete(self, code, cursor_pos=None): results = [] - ip_completer = global_dict.get('ip_completer') - p_completer = global_dict.get("p_completer") if ip_completer is not None: results = ip_completer.complete(code, cursor_pos=cursor_pos)[1] elif p_completer is not None: - p_completer = global_dict.get("p_completer") state = 0 result = None while state == 0 or result is not None: