Sfoglia il codice sorgente

HUE-4247 [batch] Support for batch pyspark or spark

Romain Rigaux 9 anni fa
parent
commit
61f0c4c

+ 81 - 2
apps/oozie/src/oozie/models2.py

@@ -17,6 +17,7 @@
 
 import json
 import logging
+import os
 import re
 import time
 import uuid
@@ -758,6 +759,17 @@ class Node():
       self.data['properties']['app_jar'] = properties['app_jar'] # Not used here
       self.data['properties']['files'] = [{'value': f['path']} for f in properties['files']]
       self.data['properties']['arguments'] = [{'value': prop} for prop in properties['arguments']]
+    elif self.data['type'] == SparkDocumentAction.TYPE:
+      notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
+      properties = notebook.get_data()['snippets'][0]['properties']
+
+      self.data['properties']['class'] = properties['class']
+      self.data['properties']['jars'] = os.path.basename(properties['jars'][0])
+      self.data['properties']['files'] = [{'value': f} for f in properties['jars']] + [{'value': f['path']} for f in properties['files']]
+      self.data['properties']['spark_arguments'] = [{'value': prop} for prop in properties['spark_arguments']]
+      self.data['properties']['spark_opts'] = ' '.join(properties['spark_opts'])
+      if len(properties['jars']) > 1:
+        self.data['properties']['spark_opts'] += ' --py-files ' + ','.join([os.path.basename(f) for f in properties['jars'][1:]])
 
     data = {
       'node': self.data,
@@ -2140,6 +2152,11 @@ class JavaDocumentAction(Action):
     return [cls.FIELDS['uuid']]
 
 
+class SparkDocumentAction(SparkAction):
+  TYPE = 'spark2-document'
+
+
+
 class DecisionNode(Action):
   TYPE = 'decision'
   FIELDS = {}
@@ -2974,8 +2991,12 @@ class WorkflowBuilder():
     for document in documents:
       if document.type == 'query-java':
         node = self.get_java_document_node(document, name)
-      else:
+      elif document.type == 'query-hive':
         node = self.get_hive_document_node(document, user)
+      elif document.type == 'query-spark2':
+        node = self.get_spark_document_node(document, user)
+      else:
+        raise PopupException(_('Snippet type %(type)s is not supported in batch execution.') % document.type)
 
       nodes.append(node)
 
@@ -3069,6 +3090,64 @@ class WorkflowBuilder():
 
     return node
 
+  def _get_spark_node(self, node_id, user, is_document_node=False):
+    credentials = []
+
+    return {
+        u'id': node_id,
+        u'name': u'spark2-%s' % node_id[:4],
+        u"type": u"spark2-document-widget", # if is_document_node else u"hive2-widget",
+        u'properties': {
+            u'files': [],
+            u'job_xml': u'',
+            u'retry_interval': [],
+            u'retry_max': [],
+            u'job_properties': [],
+            u'sla': [
+                {u'key': u'enabled', u'value': False},
+                {u'key': u'nominal-time', u'value': u'${nominal_time}'},
+                {u'key': u'should-start', u'value': u''},
+                {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
+                {u'key': u'max-duration', u'value': u''},
+                {u'key': u'alert-events', u'value': u''},
+                {u'key': u'alert-contact', u'value': u''},
+                {u'key': u'notification-msg', u'value': u''},
+                {u'key': u'upstream-apps', u'value': u''},
+            ],
+            u'archives': [],
+            u'prepares': [],
+            u'credentials': credentials,
+            u'spark_master': u'yarn',
+            u'mode': u'client',
+            u'app_name': u'BatchSpark2'
+        },
+        u'children': [
+            {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
+            {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
+        ],
+        u'actionParameters': [],
+    }
+
+  def get_spark_snippet_node(self, snippet):
+    credentials = []
+
+    node_id = snippet.get('id', str(uuid.uuid4()))
+
+    node = self._get_java_node(node_id, credentials)
+    node['properties']['class'] = snippet['properties']['class']
+    node['properties']['jars'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
+    node['properties']['spark_opts'] = [{'value': f['path']} for f in snippet['properties']['files']]
+    node['properties']['spark_arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
+
+    return node
+
+  def get_spark_document_node(self, document, user):
+    node = self._get_spark_node(document.uuid, user, is_document_node=True)
+
+    node['properties']['uuid'] = document.uuid
+
+    return node
+
   def _get_java_node(self, node_id, credentials=None, is_document_node=False):
     if credentials is None:
       credentials = []
@@ -3087,7 +3166,7 @@ class WorkflowBuilder():
               "capture_output": False,
               "prepares": [],
               "credentials": credentials,
-              "sla":[{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
+              "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
               "archives": []
         },
         "children": [

+ 1 - 0
apps/oozie/src/oozie/templates/editor2/gen/workflow-spark2-document.xml.mako

@@ -0,0 +1 @@
+workflow-spark.xml.mako

+ 4 - 0
desktop/conf.dist/hue.ini

@@ -703,6 +703,10 @@
       name=Java
       interface=oozie
 
+    [[[spark2]]]
+      name=Spark
+      interface=oozie
+
     # [[[mysql]]]
     #   name=MySql JDBC
     #   interface=jdbc

+ 4 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -711,6 +711,10 @@
       name=Java
       interface=oozie
 
+    [[[spark2]]]
+      name=Spark
+      interface=oozie
+
     # [[[mysql]]]
     #   name=MySql JDBC
     #   interface=jdbc

+ 4 - 0
desktop/libs/notebook/src/notebook/conf.py

@@ -179,6 +179,10 @@ def _default_interpreters():
           'name': 'Java', 'interface': 'oozie', 'options': {}
       })
       ,
+      ('spark2', {
+          'name': 'Spark', 'interface': 'oozie', 'options': {}
+      })
+      ,
       ('text', {
           'name': 'Text', 'interface': 'text', 'options': {}
       }),

+ 12 - 2
desktop/libs/notebook/src/notebook/static/notebook/js/notebook.ko.js

@@ -182,6 +182,15 @@
       properties['files'] = [];
     }
 
+    if (snippetType == 'spark2') {
+      properties['app_name'] = '';
+      properties['class'] = '';
+      properties['jars'] = [];
+      properties['spark_opts'] = [];
+      properties['spark_arguments'] = [];
+      properties['files'] = [];
+    }
+
     if (snippetType == 'jar' || snippetType == 'java') {
       properties['app_jar'] = '';
       properties['class'] = '';
@@ -696,8 +705,9 @@
 
     self.wasBatchExecuted = ko.observable(typeof snippet.wasBatchExecuted != "undefined" && snippet.wasBatchExecuted != null ? snippet.wasBatchExecuted : false);
     self.isReady = ko.computed(function() {
-      return ((self.type() != 'jar' && self.type() != 'java') && self.statement() !== '') ||
-        ((self.type() == 'jar' || self.type() == 'java') && (self.properties().app_jar() != '' && self.properties().class() != ''));
+      return (['jar', 'java', 'spark2'].indexOf(self.type()) == -1 && self.statement() !== '') ||
+        (['jar', 'java'].indexOf(self.type()) != -1 && (self.properties().app_jar() != '' && self.properties().class() != '')) ||
+        (['spark2'].indexOf(self.type()) != -1 && self.properties().jars().length > 0);
     });
     self.lastExecuted = ko.observable(typeof snippet.lastExecuted != "undefined" && snippet.lastExecuted != null ? snippet.lastExecuted : 0);
 

+ 26 - 5
desktop/libs/notebook/src/notebook/templates/editor_components.mako

@@ -299,11 +299,15 @@ ${ hueIcons.symbols() }
                 <i class="fa fa-file-code-o app-icon" style="vertical-align: middle"></i>
                 Java
               <!-- /ko -->
+              <!-- ko if: editorType() == 'spark2' -->
+                <img src="${ static('spark/art/icon_spark_48.png') }" class="app-icon" />
+                Spark
+              <!-- /ko -->
               <!-- ko if: editorType() == 'beeswax' || editorType() == 'hive' -->
                 <img src="${ static('beeswax/art/icon_beeswax_48.png') }" class="app-icon" />
                 Hive
               <!-- /ko -->
-              <!-- ko if: ['impala', 'pig', 'hive', 'beeswax', 'rdbms', 'java'].indexOf(editorType()) == -1 -->
+              <!-- ko if: ['impala', 'pig', 'hive', 'beeswax', 'rdbms', 'java', 'spark2'].indexOf(editorType()) == -1 -->
                 <img src="${ static('rdbms/art/icon_rdbms_48.png') }" class="app-icon" />
                 SQL
               <!-- /ko -->
@@ -1415,17 +1419,17 @@ ${ hueIcons.symbols() }
                 <!-- ko template: { if: $root.editorMode(), name: 'editor-snippet-header' } --><!-- /ko -->
                 <!-- ko template: { if: ! $root.editorMode(), name: 'notebook-snippet-header' } --><!-- /ko -->
               </h2>
-              <!-- ko template: { if: ['text', 'jar', 'java', 'py', 'markdown'].indexOf(type()) == -1, name: 'code-editor-snippet-body' } --><!-- /ko -->
+              <!-- ko template: { if: ['text', 'jar', 'java', 'spark2', 'py', 'markdown'].indexOf(type()) == -1, name: 'code-editor-snippet-body' } --><!-- /ko -->
               <!-- ko template: { if: type() == 'text', name: 'text-snippet-body' } --><!-- /ko -->
               <!-- ko template: { if: type() == 'markdown', name: 'markdown-snippet-body' } --><!-- /ko -->
-              <!-- ko template: { if: ['java', 'jar', 'py'].indexOf(type()) != -1, name: 'executable-snippet-body' } --><!-- /ko -->
+              <!-- ko template: { if: ['java', 'jar', 'py', 'spark2'].indexOf(type()) != -1, name: 'executable-snippet-body' } --><!-- /ko -->
             </div>
             <div style="position: absolute; top:25px; margin-left:35px; width: calc(100% - 35px)" data-bind="style: { 'z-index': 400 - $index() }">
               <!-- ko template: 'snippet-settings' --><!-- /ko -->
             </div>
           </div>
           <!-- ko template: { if: ['text', 'markdown'].indexOf(type()) == -1, name: 'snippet-execution-status' } --><!-- /ko -->
-          <!-- ko template: { if: $root.editorMode() && type() != 'java', name: 'snippet-code-resizer' } --><!-- /ko -->
+          <!-- ko template: { if: $root.editorMode() && ['java', 'spark2'].indexOf(type()) == -1, name: 'snippet-code-resizer' } --><!-- /ko -->
           <!-- ko if: $root.editorMode() -->
           <!-- ko template: 'snippet-log' --><!-- /ko -->
           <!-- ko template: 'query-tabs' --><!-- /ko -->
@@ -1806,7 +1810,24 @@ ${ hueIcons.symbols() }
         </div>
       </div>
       <!-- /ko -->
-      <!-- ko template: { if: typeof properties().arguments != 'undefined', name: 'property', data: { type: 'csv', label: '${ _ko('Arguments') }', value: properties().arguments, title: '${ _ko('The YARN queue to submit to (Default: default)') }', placeholder: '${ _ko('e.g. -foo=bar') }', inline: false }} --><!-- /ko -->
+      <!-- ko if: type() == 'spark2' -->
+      <div class="control-group">
+        <!-- ko template: { if: typeof properties().jars != 'undefined', name: 'property', data: { type: 'csv-hdfs-files', label: '${ _ko('Libs') }', value: properties().jars, title: '${ _ko('Path to jar or python files.') }', placeholder: '${ _ko('e.g. /user/hue/pi.py') }'}} --><!-- /ko -->
+      </div>
+      <!-- ko if: $.grep(properties().jars(), function(val, index) { return val.toLowerCase().endsWith('.jar'); }).length > 0 -->
+        <div class="control-group">
+          <label class="control-label">
+            ${_('Class')}
+          </label>
+          <div class="controls">
+            <input type="text" class="input-xxlarge" data-bind="value: properties().class, valueUpdate: 'afterkeydown'" placeholder="${ _('Class name of application, e.g. org.apache.oozie.example.SparkFileCopy') }"/>
+          </div>
+        </div>
+      <!-- /ko -->
+      <div class="control-group">
+        <!-- ko template: { if: typeof properties().spark_arguments != 'undefined', name: 'property', data: { type: 'csv-hdfs-files', label: '${ _ko('Arguments') }', value: properties().spark_arguments, title: '${ _ko('Arguments to the application.') }', placeholder: '${ _ko('e.g. 10, /user/hue/input') }'}} --><!-- /ko -->
+      </div>
+      <!-- /ko -->
     </form>
   </div>
 </script>