Enormous refactor of Livyclientlib

remotespark/example_config.json

@@ -1,10 +1,4 @@
 {
-  "serialize": false,
-  "serialize_periodically": false,
-  "serialize_period_seconds": 3,
-
-  "default_chart_type": "area",
-
   "kernel_python_credentials" : {
     "username": "",
     "password": "",
@@ -40,10 +34,10 @@
     }
   },
 
-  "execute_timeout_seconds": 3600,
+  "wait_for_idle_timeout_seconds": 15,
   "status_sleep_seconds": 2,
   "statement_sleep_seconds": 2,
-  "create_sql_context_timeout_seconds": 60,
+  "livy_session_startup_timeout_seconds": 60,
 
   "fatal_error_suggestion": "The code failed because of a fatal error:\n\t{}.\n\nSome things to try:\na) Make sure Spark has enough available resources for Jupyter to create a Spark context.\nb) Contact your Jupyter administrator to make sure the Spark magics library is configured correctly.\nc) Restart the kernel.",
 

remotespark/kernels/kernelmagics.py

@@ -13,6 +13,7 @@
 from IPython.core.magic_arguments import argument, magic_arguments, parse_argstring
 
 import remotespark.utils.configuration as conf
+from remotespark.livyclientlib.command import Command
 from remotespark.livyclientlib.sqlquery import SQLQuery
 from remotespark.magics.sparkmagicsbase import SparkMagicBase
 from remotespark.utils.constants import LANGS_SUPPORTED
@@ -155,7 +156,7 @@ def configure(self, line, cell="", local_ns=None):
     @cell_magic
     def spark(self, line, cell="", local_ns=None):
         if self._do_not_call_start_session(""):
-            (success, out) = self.spark_controller.run_cell(cell)
+            (success, out) = self.spark_controller.run_command(Command(cell))
             if success:
                 self.ipython_display.write(out)
             else:

remotespark/livyclientlib/command.py

@@ -0,0 +1,49 @@
+import textwrap
+
+from remotespark.utils.guid import ObjectWithGuid
+from remotespark.utils.log import Log
+
+
+class Command(ObjectWithGuid):
+    def __init__(self, code):
+        super(Command, self).__init__()
+        self.code = textwrap.dedent(code)
+        self.logger = Log("Command")
+
+    def __eq__(self, other):
+        return self.code == other.code
+
+    def __ne__(self, other):
+        return not self == other
+
+    def execute(self, session):
+        session.wait_for_idle()
+        data = {"code": self.code}
+        response = session.http_client.post_statement(session.id, data)
+        statement_id = response['id']
+        return self._get_statement_output(session, statement_id)
+
+    def _get_statement_output(self, session, statement_id):
+        statement_running = True
+        out = ""
+        while statement_running:
+            statement = session.http_client.get_statement(session.id, statement_id)
+            status = statement["state"]
+
+            self.logger.debug("Status of statement {} is {}.".format(statement_id, status))
+
+            if status == "running":
+                session.sleep()
+            else:
+                statement_running = False
+
+                statement_output = statement["output"]
+                if statement_output["status"] == "ok":
+                    out = (True, statement_output["data"]["text/plain"])
+                elif statement_output["status"] == "error":
+                    out = (False,
+                           statement_output["evalue"] + "\n" + "".join(statement_output["traceback"]))
+                else:
+                    raise ValueError("Unknown output status: '{}'".format(statement_output["status"]))
+
+        return out