Browse Source

HUE-2677 [spark] Implement Livy batches API for jars

Romain Rigaux 10 years ago
parent
commit
e3fed26ffd

+ 13 - 1
apps/spark/src/spark/job_server_api.py

@@ -79,7 +79,7 @@ class JobServerApi(object):
       self._thread_local.user = user.username
     else:
       self._thread_local.user = user
-  
+
   def get_status(self):
     return self._root.get('sessions')
 
@@ -102,3 +102,15 @@ class JobServerApi(object):
 
   def cancel(self, session):
     return self._root.post('sessions/%s/interrupt' % session)
+
+  def get_batches(self):
+    return self._root.get('batches')
+
+  def submit_batch(self, properties):
+    return self._root.post('batches', data=json.dumps(properties), contenttype=_JSON_CONTENT_TYPE)
+
+  def get_batch(self, uuid):
+    return self._root.get('batches/%s' % uuid)
+
+  def delete_batch(self, uuid):
+    return self._root.delete('batches/%s' % uuid)

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

@@ -81,8 +81,10 @@ class Notebook():
 
 
 def get_api(user, snippet):
-  if snippet['type'] in ('hive', 'impala', 'spark-sql', 'jar'):
+  if snippet['type'] in ('hive', 'impala', 'spark-sql'):
     return HS2Api(user)
+  elif snippet['type'] == 'jar':
+    return SparkBatchApi(user)
   elif snippet['type'] == 'text':
     return TextApi(user)
   else:
@@ -171,7 +173,7 @@ class HS2Api():
     db = self._get_db(snippet)
 
     handle = self._get_handle(snippet)
-    operation =  db.get_operation_status(handle)
+    operation = db.get_operation_status(handle)
     status = HiveServerQueryHistory.STATE_MAP[operation.operationState]
 
     if status.index in (QueryHistory.STATE.failed.index, QueryHistory.STATE.expired.index):
@@ -294,7 +296,7 @@ class SparkApi():
     session = _get_snippet_session(notebook, snippet)
 
     try:
-      response = api.submit_statement(session['id'], snippet['statement']) ## --> post all props in json
+      response = api.submit_statement(session['id'], snippet['statement'])
       return {
           'id': response['id'],
           'has_result_set': True,
@@ -405,3 +407,57 @@ class SparkApi():
 
   def _get_jobs(self, log):
     return []
+
+
+class SparkBatchApi():
+
+  def __init__(self, user):
+    self.user = user
+
+  def create_session(self, lang):
+    return {
+        'type': lang,
+        'id': None
+    }
+
+  def execute(self, notebook, snippet):
+    api = get_spark_api(self.user)
+
+    properties = {
+        'file': snippet['properties']['app_jar'],
+        'className': snippet['properties']['class'],
+        'args': snippet['properties']['arguments'],
+        # pyFiles
+        # files
+        # driverMemory
+        # driverCores
+        # executorMemory
+        # executorCores
+        # archives
+    }
+
+    response = api.submit_batch(properties)
+    return {
+        'id': response['id'],
+        'has_result_set': True,
+    }
+
+  def check_status(self, notebook, snippet):
+    api = get_spark_api(self.user)
+
+    response = api.get_batch(snippet['result']['handle']['id'])
+    return {
+        'status': response['state'],
+    }
+
+  def get_log(self, snippet):
+    api = get_spark_api(self.user)
+
+    response = api.get_batch(snippet['result']['handle']['id'])
+    return '\n'.join(response['lines'])
+
+  def _progress(self, snippet, logs):
+    return 50
+
+  def _get_jobs(self, log):
+    return []

+ 1 - 0
apps/spark/src/spark/static/spark/js/spark.ko.js

@@ -276,6 +276,7 @@ var Snippet = function (vm, notebook, snippet) {
       type: self.type,
       status: self.status,
       statement: self.statement,
+      properties: self.properties,
       result: self.result.getContext()
     };
   }

+ 1 - 1
desktop/conf.dist/hue.ini

@@ -1034,7 +1034,7 @@
 
 [spark]
   # URL of the REST Spark Job Server.
-  ## server_url=http://localhost:8090/
+  ## server_url=http://localhost:8998/
 
   # List of available types of snippets
   ## languages='[{"name": "Scala", "type": "scala"},{"name": "Python", "type": "python"},{"name": "Impala SQL", "type": "impala"},{"name": "Hive SQL", "type": "hive"},{"name": "Text", "type": "text"}]'

+ 1 - 1
desktop/conf/pseudo-distributed.ini.tmpl

@@ -1041,7 +1041,7 @@
 
 [spark]
   # URL of the REST Spark Job Server.
-  ## server_url=http://localhost:8090/
+  ## server_url=http://localhost:8998/
 
   # List of available types of snippets
   ## languages='[{"name": "Scala", "type": "scala"},{"name": "Python", "type": "python"},{"name": "Impala SQL", "type": "impala"},{"name": "Hive SQL", "type": "hive"},{"name": "Text", "type": "text"}]'