فهرست منبع

[spark] Move createSession() to the Notebook class

Harmonize how we create session and close them.
This will support creating default session and new session with specific
properties.
Even if the Notebook take care of creating session, we can still
update the status of each snippet.
Romain Rigaux 10 سال پیش
والد
کامیت
5e704d5

+ 9 - 7
apps/spark/src/spark/api.py

@@ -40,15 +40,17 @@ def create_session(request):
   response = {'status': -1}
 
   notebook = json.loads(request.POST.get('notebook', '{}'))
-  snippet = json.loads(request.POST.get('snippet', '{}'))
+  session = json.loads(request.POST.get('session', '{}'))
 
-  session = [session for session in notebook['sessions'] if snippet['type'] == session['type']]
-  if any(session) and 'properties' in session[0]:
-    properties = session[0]['properties']
-  else:
-    properties = None
+  properties = session.get('properties', [])
+
+  # If not properties look for previously used notebook session
+  if not properties:
+    old_session = [_session for _session in notebook['sessions'] if _session['type'] == session['type']]
+    if any(old_session) and 'properties' in old_session[0]:
+      properties = old_session[0]['properties']
 
-  response['session'] = get_api(request.user, snippet).create_session(lang=snippet['type'], properties=properties)
+  response['session'] = get_api(request.user, session).create_session(lang=session['type'], properties=properties)
   response['status'] = 0
 
   return JsonResponse(response)

+ 3 - 10
apps/spark/src/spark/models.py

@@ -160,13 +160,6 @@ class HS2Api(Api):
 
     return dbms.get(self.user, query_server=get_query_server_config(name=name))
 
-  def create_session(self, lang, properties=None):
-    return {
-        'type': lang,
-        'id': None, # Real one at some point
-        'properties': []
-    }
-
   def execute(self, notebook, snippet):
     db = self._get_db(snippet)
     query = hql_query(snippet['statement'], QUERY_TYPES[0])
@@ -295,14 +288,14 @@ class SparkApi(Api):
     {'name': 'jars', 'nice_name': _('Jars'), 'default': '', 'type': 'csv', 'is_yarn': False},
     {'name': 'files', 'nice_name': _('Files'), 'default': '', 'type': 'csv', 'is_yarn': False},
     {'name': 'pyFiles', 'nice_name': _('pyFiles'), 'default': '', 'type': 'csv', 'is_yarn': False},
-    
+
     {'name': 'driverMemory', 'nice_name': _('Driver Memory'), 'default': '1', 'type': 'jvm', 'is_yarn': False},
-    
+
     {'name': 'driverCores', 'nice_name': _('Driver Cores'), 'default': '1', 'type': 'number', 'is_yarn': True},
     {'name': 'executorCores', 'nice_name': _('Executor Cores'), 'default': '1', 'type': 'number', 'is_yarn': True},
     {'name': 'queue', 'nice_name': _('Queue'), 'default': '1', 'type': 'string', 'is_yarn': True},
     {'name': 'archives', 'nice_name': _('Archives'), 'default': '', 'type': 'csv', 'is_yarn': True},
-    {'name': 'numExecutors', 'nice_name': _('Executors Numbers'), 'default': '1', 'type': 'number', 'is_yarn': True},    
+    {'name': 'numExecutors', 'nice_name': _('Executors Numbers'), 'default': '1', 'type': 'number', 'is_yarn': True},
   ]
 
   def create_session(self, lang='scala', properties=None):

+ 52 - 64
apps/spark/src/spark/static/spark/js/spark.ko.js

@@ -314,9 +314,9 @@ var Snippet = function (vm, notebook, snippet) {
     };
   }
 
-  self._ajax_error = function (data, callback) {
+  self._ajaxError = function (data, callback) {
     if (data.status == -2) {
-      self.create_session(callback);
+      notebook.createSession({'type': self.type()}, callback);
     }
     else if (data.status == -3) {
       self.status('expired');
@@ -330,34 +330,6 @@ var Snippet = function (vm, notebook, snippet) {
     }
   };
 
-  self.create_session = function (callback, errorCallback) {
-    self.status('loading');
-    $.post("/spark/api/create_session", {
-      notebook: ko.mapping.toJSON(notebook.getContext()),
-      snippet: ko.mapping.toJSON(self.getContext())
-    }, function (data) {
-      if (data.status == 0) {
-        notebook.addSession(ko.mapping.fromJS(data.session));
-        self.status('ready');
-        if (callback) {
-          setTimeout(callback, 500);
-        }
-      }
-      else {
-        self.status('failed');
-        $(document).trigger("error", data.message);
-        if (errorCallback) {
-          errorCallback();
-        }
-      }
-    }).fail(function (xhr, textStatus, errorThrown) {
-      $(document).trigger("error", xhr.responseText);
-      if (errorCallback) {
-        errorCallback();
-      }
-    });
-  };
-
   self.execute = function () {
     if (self.status() == 'running' || self.status() == 'loading') {
       return;
@@ -387,7 +359,7 @@ var Snippet = function (vm, notebook, snippet) {
         self.result.hasResultset(data.handle.has_result_set);
         self.checkStatus();
       } else {
-        self._ajax_error(data, self.execute);
+        self._ajaxError(data, self.execute);
       }
     }).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
@@ -436,7 +408,7 @@ var Snippet = function (vm, notebook, snippet) {
           }, 500);
         }
       } else {
-        self._ajax_error(data);
+        self._ajaxError(data);
         $(document).trigger("renderDataError", {snippet: self});
       }
     }).fail(function (xhr, textStatus, errorThrown) {
@@ -481,7 +453,7 @@ var Snippet = function (vm, notebook, snippet) {
           self.progress(99);
         }
       } else {
-        self._ajax_error(data);
+        self._ajaxError(data);
       }
     }).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
@@ -502,7 +474,7 @@ var Snippet = function (vm, notebook, snippet) {
       if (data.status == 0) {
         self.status('canceled');
       } else {
-        self._ajax_error(data);
+        self._ajaxError(data);
       }
     }).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
@@ -523,7 +495,7 @@ var Snippet = function (vm, notebook, snippet) {
       if (data.status == 0) {
         // self.status('closed'); // Keep as 'running' as currently it happens before running a new query
       } else {
-        self._ajax_error(data);
+        self._ajaxError(data);
       }
     }).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
@@ -550,7 +522,7 @@ var Snippet = function (vm, notebook, snippet) {
         }
         self.progress(data.progress);
       } else {
-        self._ajax_error(data);
+        self._ajaxError(data);
       }
     }).fail(function (xhr, textStatus, errorThrown) {
       $(document).trigger("error", xhr.responseText);
@@ -594,28 +566,8 @@ var Notebook = function (vm, notebook) {
   };
 
   self.restartSession = function (session) {
-    var snippetsOfType = $.grep(self.snippets(), function (snippet) {
-      return snippet.type() == session.type();
-    });
-
-    if (snippetsOfType.length > 0) {
-      var setSnippetStatus = function (status) {
-        $.each(snippetsOfType, function (index, snippet) {
-          snippet.status(status);
-        });
-      };
-
-      self.closeSession(session);
-
-      setSnippetStatus('loading');
-      var successCallback = function() {
-        setSnippetStatus('ready');
-      };
-      var failCallback = function() {
-        setSnippetStatus('failed');
-      };
-      snippetsOfType[0].create_session(successCallback, failCallback);
-    }
+    self.closeSession(session);
+    self.createSession(session);
   };
 
   self.addSession = function (session) {
@@ -638,7 +590,7 @@ var Notebook = function (vm, notebook) {
     self.snippets.push(_snippet);
 
     if (self.getSession(_snippet.type()) == null) {
-      _snippet.create_session();
+      self.createSession({'type': _snippet.type()});
     }
 
     _snippet.init();
@@ -683,8 +635,8 @@ var Notebook = function (vm, notebook) {
     });
     self.snippets.push(_snippet);
 
-    if (self.getSession(self.selectedSnippet()) == null) {
-      _snippet.create_session();
+    if (self.getSession(_snippet.type()) == null) {
+      self.createSession({'type': _snippet.type()});
     }
     else {
       _snippet.status('ready');
@@ -741,6 +693,42 @@ var Notebook = function (vm, notebook) {
     });
   };
 
+  self.createSession = function (session, callback) {
+    var snippets = $.grep(self.snippets(), function (snippet) {
+       return snippet.type() == session.type;
+    });
+
+    $.each(snippets, function(index, snippet) {
+      snippet.status('loading');
+    });
+
+    $.post("/spark/api/create_session", {
+      notebook: ko.mapping.toJSON(self.getContext()),
+      session: ko.mapping.toJSON(session), // e.g. {'type': 'hive', 'properties': [{'driverCores': '2'}]}
+    }, function (data) {
+      if (data.status == 0) {
+        self.addSession(ko.mapping.fromJS(data.session));
+        $.each(snippets, function(index, snippet) {
+          snippet.status('ready');
+        });
+        if (callback) {
+          setTimeout(callback, 500);
+        }
+      }
+      else {
+        $.each(snippets, function(index, snippet) {
+          snippet.status('failed');
+    	});
+        $(document).trigger("error", data.message);
+      }
+    }).fail(function (xhr, textStatus, errorThrown) {
+      $.each(snippets, function(index, snippet) {
+        snippet.status('failed');
+      })
+      $(document).trigger("error", xhr.responseText);
+    });
+  };
+
   self.closeSession = function (session) {
     $.post("/spark/api/close_session", {
       session: ko.mapping.toJSON(session)
@@ -766,7 +754,7 @@ function EditorViewModel(notebooks, options) {
   self.combinedContent = ko.observable();
 
   self.displayCombinedContent = function () {
-    if (!self.selectedNotebook()) {
+    if (! self.selectedNotebook()) {
       self.combinedContent('');
     } else {
       var statements = '';
@@ -802,7 +790,7 @@ function EditorViewModel(notebooks, options) {
 
   self.availableSnippets = ko.mapping.fromJS(options.languages);
   self.snippetPlaceholders = options.snippet_placeholders;
- 
+
   self.availableSessionProperties = ko.computed(function () { // Only Spark
     return ko.utils.arrayFilter(options.session_properties, function (item) {
         return item.name != '' // Could filter out the ones already selected + yarn only or not
@@ -816,7 +804,7 @@ function EditorViewModel(notebooks, options) {
         _prop = prop;
         return;
       }
-    });console.log(_prop);
+    });
     return _prop;
   };
 

+ 2 - 2
apps/spark/src/spark/templates/editor_components.mako

@@ -796,10 +796,10 @@ from django.utils.translation import ugettext as _
                     <!-- /ko -->
                     <!-- ko if: type == 'string' -->
                     <input class="input-small" type="text" data-bind="value: $parent.value" />
-                    <!-- /ko -->                  
+                    <!-- /ko -->
                     <!-- ko if: type == 'csv' -->
                     <input class="input-small" type="text" data-bind="value: $parent.value" />
-                    <!-- /ko -->                  
+                    <!-- /ko -->
                   <!-- /ko -->
                   <a href="javascript:void(0)" data-bind="click: function(data) { $parent.properties.remove(data) }">
                     <i class="fa fa-minus"></i>