Browse Source

HUE-4526 [indexer] Submit notebook as batch job

Romain Rigaux 9 years ago
parent
commit
8042ff02c6

+ 63 - 23
apps/oozie/src/oozie/models2.py

@@ -2962,10 +2962,9 @@ class WorkflowBuilder():
   Building a workflow that has saved Documents for nodes (e.g Saved Hive query, saved Pig script...).
   Building a workflow that has saved Documents for nodes (e.g Saved Hive query, saved Pig script...).
   """
   """
 
 
-  def create_workflow(self, user, document=None, documents=None, name=None, managed=False):
+  def create_workflow(self, user, document=None, name=None, managed=False):
     nodes = []
     nodes = []
-    if documents is None:
-      documents = [document]
+    documents = [document]
 
 
     if name is None:
     if name is None:
       name = _('Schedule of ') + ','.join([document.name or document.type for document in documents])
       name = _('Schedule of ') + ','.join([document.name or document.type for document in documents])
@@ -2984,6 +2983,28 @@ class WorkflowBuilder():
     return workflow_doc
     return workflow_doc
 
 
 
 
+  def create_notebook_workflow(self, user, notebook=None, name=None, managed=False):
+    nodes = []
+
+    if name is None:
+      name = _('Schedule of ') + ','.join([snippet['name'] or snippet['type'] for snippet in notebook['snippets']])
+
+    for snippet in notebook['snippets']:
+      print snippet
+      if snippet['type'] == 'java':
+        node = self.get_java_snippet_node(snippet)
+      elif snippet['type'] == 'java':
+        node = self.get_hive_document_node(snippet, user)
+      else:
+        raise PopupException(_('Snippet type %(type)s is not supported in batch execution.') % snippet)
+
+      nodes.append(node)
+
+    workflow_doc = self.get_workflow(nodes, name, notebook['uuid'], user, managed=managed) # TODO optionally save
+
+    return workflow_doc
+
+
   def get_hive_document_node(self, document, name, user):
   def get_hive_document_node(self, document, name, user):
     api = get_oozie(user)
     api = get_oozie(user)
 
 
@@ -3000,7 +3021,7 @@ class WorkflowBuilder():
         u'properties': {
         u'properties': {
             u'files': [],
             u'files': [],
             u'job_xml': u'',
             u'job_xml': u'',
-            u'uuid': document.uuid,
+            u'uuid': document.uuid, # + snippet uuid
             u'parameters': parameters,
             u'parameters': parameters,
             u'retry_interval': [],
             u'retry_interval': [],
             u'retry_max': [],
             u'retry_max': [],
@@ -3024,41 +3045,60 @@ class WorkflowBuilder():
         },
         },
         u'children': [
         u'children': [
             {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
             {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
-            {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'
-        }],
+            {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
+        ],
         u'actionParameters': [],
         u'actionParameters': [],
     }
     }
 
 
-  def get_java_document_node(self, document, name):
-    credentials = []
+  def _get_java_node(self, node_id, credentials=None, is_document_node=False):
+    if credentials is None:
+      credentials = []
 
 
     return {
     return {
-        "id": str(uuid.uuid4()),
-        'name': u'doc-hive-%s' % document.uuid[:4],
-        "type":"java-document-widget",
+        "id": node_id,
+        'name': 'doc-hive-%s' % node_id[:4],
+        "type": "java-document-widget" if is_document_node else "java-widget",
         "properties":{
         "properties":{
-              u'uuid': document.uuid, # Files, main_class, arguments comes from there
-              "job_xml":[],
+              "job_xml": [],
               "jar_path": "",
               "jar_path": "",
-              "java_opts":[],
-              "retry_max":[],
-              "retry_interval":[],
-              "job_properties":[],
+              "java_opts": [],
+              "retry_max": [],
+              "retry_interval": [],
+              "job_properties": [],
               "capture_output": False,
               "capture_output": False,
-              "prepares":[],
+              "prepares": [],
               "credentials": credentials,
               "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":[]
+              "archives": []
         },
         },
-        "children":[
-            {"to":"33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
-            {"error":"17c9c895-5a16-7443-bb81-f34b30b21548"}
+        "children": [
+            {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
+            {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
         ],
         ],
-        "actionParameters":[],
+        "actionParameters": [],
         "actionParametersFetched": False
         "actionParametersFetched": False
     }
     }
 
 
+  def get_java_snippet_node(self, snippet):
+    credentials = []
+
+    node_id = snippet.get('id') or str(uuid.uuid4())
+
+    node = self._get_java_node(node_id, credentials)
+    node['properties']['main_class'] = snippet['properties']['class']
+    node['properties']['app_jar'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
+    node['properties']['files'] = [{'value': f['path']} for f in snippet['properties']['files']]
+    node['properties']['arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
+
+    return node
+
+  def get_java_document_node(self, document):
+    credentials = []
+
+    node = self._get_java_node(document.uuid, credentials, is_document_node=True)
+    node['uuid'] = document.uuid
 
 
+    return node
 
 
   def get_workflow(self, nodes, name, doc_uuid, user, managed=False):
   def get_workflow(self, nodes, name, doc_uuid, user, managed=False):
     parameters = []
     parameters = []

+ 8 - 7
desktop/libs/indexer/src/indexer/api3.py

@@ -93,8 +93,8 @@ def guess_field_types(request):
 
 
     format_ = indexer.guess_field_types({
     format_ = indexer.guess_field_types({
       "file": {
       "file": {
-        "stream": stream,
-        "name": file_format['path']
+          "stream": stream,
+          "name": file_format['path']
         },
         },
       "format": file_format['format']
       "format": file_format['format']
     })
     })
@@ -111,8 +111,9 @@ def guess_field_types(request):
         ]
         ]
     }
     }
   elif file_format['inputFormat'] == 'query':
   elif file_format['inputFormat'] == 'query':
-    #TODO get schema from explain query
-    pass
+    # TODO get schema from explain query, which is not possible
+    # Only support select * for now and would require a workflow that generates the morphline on the fly based on a temporary CTAS
+    format_ = {u'sample': [[u'00-0000', u'All Occupations', 134354250, 40690], [u'11-0000', u'Management occupations', 6003930, 96150], [u'11-1011', u'Chief executives', 299160, 151370], [u'11-1021', u'General and operations managers', 1655410, 103780]], u'columns': [{u'operations': [], u'name': u'code', u'required': False, u'keep': True, u'unique': False, u'type': u'string'}, {u'operations': [], u'name': u'description', u'required': False, u'keep': True, u'unique': False, u'type': u'string'}, {u'operations': [], u'name': u'total_emp', u'required': False, u'keep': True, u'unique': False, u'type': u'string'}, {u'operations': [], u'name': u'salary', u'required': False, u'keep': True, u'unique': False, u'type': u'string'}]}
 
 
   return JsonResponse(format_)
   return JsonResponse(format_)
 
 
@@ -139,8 +140,8 @@ def index_file(request):
     db = dbms.get(request.user)
     db = dbms.get(request.user)
     table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
     table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
     input_path = table_metadata.path_location
     input_path = table_metadata.path_location
-  else:
-    input_path = file_format["path"]
+  elif file_format['inputFormat'] == 'file':
+    input_path = '${nameNode}%s' % file_format["path"]
 
 
-  job_handle = indexer.run_morphline(request, collection_name, morphline, input_path) #TODO if query generate insert
+  job_handle = indexer.run_morphline(request, collection_name, morphline, input_path)
   return JsonResponse(job_handle)
   return JsonResponse(job_handle)

+ 39 - 68
desktop/libs/indexer/src/indexer/smart_indexer.py

@@ -23,8 +23,8 @@ from mako.lookup import TemplateLookup
 from mako.template import Template
 from mako.template import Template
 
 
 from collections import deque
 from collections import deque
-from notebook.api import _save_notebook, _execute_notebook
-from notebook.models import make_notebook, make_notebook2
+from notebook.api import _execute_notebook
+from notebook.models import make_notebook2
 from oozie.models2 import Job
 from oozie.models2 import Job
 
 
 from indexer.fields import get_field_type
 from indexer.fields import get_field_type
@@ -64,69 +64,40 @@ class Indexer(object):
   def run_morphline(self, request, collection_name, morphline, input_path):
   def run_morphline(self, request, collection_name, morphline, input_path):
     workspace_path = self._upload_workspace(morphline)
     workspace_path = self._upload_workspace(morphline)
 
 
-#     snippets = [
-#       {
-#         u'type': u'java',
-#         u'files': [
-#             {u'path': u'%s/log4j.properties' % workspace_path, u'type': u'file'},
-#             {u'path': u'%s/morphline.conf' % workspace_path, u'type': u'file'}
-#         ],
-#         u'class': u'org.apache.solr.hadoop.MapReduceIndexerTool',
-#         u'app_jar': CONFIG_INDEXER_LIBS_PATH.get(),
-#         u'arguments': [
-#             u'--morphline-file',
-#             u'morphline.conf',
-#             u'--output-dir',
-#             u'${nameNode}/user/%s/indexer' % self.username,
-#             u'--log4j',
-#             u'log4j.properties',
-#             u'--go-live',
-#             u'--zk-host',
-#             zkensemble(),
-#             u'--collection',
-#             collection_name,
-#             input_path,
-#         ],
-#         u'archives': [],
-#       }
-#     ]
-#
-#     # managed notebook
-#     notebook = make_notebook2(name='Indexer job for %s' % collection_name, snippets=snippets).get_data()
-#     notebook_doc, created = _save_notebook(notebook, self.user)
-#
-#     snippet = {'wasBatchExecuted': True}
-
-    snippet_properties =  {
-       u'files': [
-           {u'path': u'%s/log4j.properties' % workspace_path, u'type': u'file'},
-           {u'path': u'%s/morphline.conf' % workspace_path, u'type': u'file'}
-       ],
-       u'class': u'org.apache.solr.hadoop.MapReduceIndexerTool',
-       u'app_jar': CONFIG_INDEXER_LIBS_PATH.get(),
-       u'arguments': [
-           u'--morphline-file',
-           u'morphline.conf',
-           u'--output-dir',
-           u'${nameNode}/user/%s/indexer' % self.username,
-           u'--log4j',
-           u'log4j.properties',
-           u'--go-live',
-           u'--zk-host',
-           zkensemble(),
-           u'--collection',
-           collection_name,
-           input_path,
-       ],
-       u'archives': [],
-    }
-
-    notebook = make_notebook(name='Indexer', editor_type='java', snippet_properties=snippet_properties, status='running').get_data()
-    notebook_doc, created = _save_notebook(notebook, self.user)
-
-    snippet = {'wasBatchExecuted': True, 'id': notebook['snippets'][0]['id'], 'statement': ''}
-
-    job_handle = _execute_notebook(request, notebook, snippet)
+    snippets = [
+      {
+        u'type': u'java',
+        u'status': u'running',
+        u'properties':  {
+          u'files': [
+              {u'path': u'%s/log4j.properties' % workspace_path, u'type': u'file'},
+              {u'path': u'%s/morphline.conf' % workspace_path, u'type': u'file'}
+          ],
+          u'class': u'org.apache.solr.hadoop.MapReduceIndexerTool',
+          u'app_jar': CONFIG_INDEXER_LIBS_PATH.get(),
+          u'arguments': [
+              u'--morphline-file',
+              u'morphline.conf',
+              u'--output-dir',
+              u'${nameNode}/user/%s/indexer' % self.username,
+              u'--log4j',
+              u'log4j.properties',
+              u'--go-live',
+              u'--zk-host',
+              zkensemble(),
+              u'--collection',
+              collection_name,
+              input_path,
+          ],
+          u'archives': [],
+        }
+      }
+    ]
+
+    notebook = make_notebook2(name='Indexer job for %s' % collection_name, snippets=snippets).get_data()
+    snippet = {'wasBatchExecuted': True, 'type': 'oozie', 'id': notebook['snippets'][0]['id'], 'statement': ''}
+
+    job_handle = _execute_notebook(request, notebook, snippet) # To set as managed
 
 
     return job_handle
     return job_handle
 
 
@@ -220,11 +191,11 @@ class Indexer(object):
     grok_dicts_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "grok_dictionaries")
     grok_dicts_loc = os.path.join(CONFIG_INDEXER_LIBS_PATH.get(), "grok_dictionaries")
 
 
     properties = {
     properties = {
-      "collection_name":collection_name,
-      "fields":self.get_field_list(data['columns']),
+      "collection_name": collection_name,
+      "fields": self.get_field_list(data['columns']),
       "num_base_fields": len(data['columns']),
       "num_base_fields": len(data['columns']),
       "uuid_name" : uuid_name,
       "uuid_name" : uuid_name,
-      "get_regex":Indexer._get_regex_for_type,
+      "get_regex": Indexer._get_regex_for_type,
       "format_settings": data['format'],
       "format_settings": data['format'],
       "format_class": get_file_format_class(data['format']['type']),
       "format_class": get_file_format_class(data['format']['type']),
       "get_kept_args": get_checked_args,
       "get_kept_args": get_checked_args,

+ 3 - 3
desktop/libs/indexer/src/indexer/templates/indexer.mako

@@ -183,7 +183,7 @@ ${ assist.assistPanel() }
 
 
     <ol class="list-inline text-center step-indicator">
     <ol class="list-inline text-center step-indicator">
       <li data-bind="css: { 'active': currentStep() == 1, 'complete': currentStep() > 1 }, click: function() { currentStep(1) }">
       <li data-bind="css: { 'active': currentStep() == 1, 'complete': currentStep() > 1 }, click: function() { currentStep(1) }">
-        <div class="step">
+        <div class="step" title="${ _('Go to Step 1') }">
           <!-- ko if: currentStep() == 1 -->
           <!-- ko if: currentStep() == 1 -->
             <!-- ko if: createWizard.isGuessingFormat -->
             <!-- ko if: createWizard.isGuessingFormat -->
               <span class="fa fa-spinner fa-spin"></span>
               <span class="fa fa-spinner fa-spin"></span>
@@ -199,7 +199,7 @@ ${ assist.assistPanel() }
         <div class="caption">${ _('Pick it') }</div>
         <div class="caption">${ _('Pick it') }</div>
       </li>
       </li>
       <li data-bind="css: { 'inactive': currentStep() == 1, 'active': currentStep() == 2, 'complete': currentStep() == 3 }, click: function() { currentStep(2) }">
       <li data-bind="css: { 'inactive': currentStep() == 1, 'active': currentStep() == 2, 'complete': currentStep() == 3 }, click: function() { currentStep(2) }">
-        <div class="step">
+        <div class="step" title="${ _('Go to Step 2') }">
           <!-- ko if: currentStep() < 3 -->
           <!-- ko if: currentStep() < 3 -->
             <!-- ko if: createWizard.isGuessingFieldTypes -->
             <!-- ko if: createWizard.isGuessingFieldTypes -->
               <span class="fa fa-spinner fa-spin"></span>
               <span class="fa fa-spinner fa-spin"></span>
@@ -215,7 +215,7 @@ ${ assist.assistPanel() }
         <div class="caption">${ _('Tweak it') }</div>
         <div class="caption">${ _('Tweak it') }</div>
       </li>
       </li>
       <li data-bind="css: { 'inactive': currentStep() < 3, 'active': currentStep() == 3, 'error': createWizard.indexingError, 'complete': createWizard.indexingSuccess }, click: function() { currentStep(3) }">
       <li data-bind="css: { 'inactive': currentStep() < 3, 'active': currentStep() == 3, 'error': createWizard.indexingError, 'complete': createWizard.indexingSuccess }, click: function() { currentStep(3) }">
-        <div class="step">
+        <div class="step" title="${ _('Go to Step 3') }">
           <!-- ko if: createWizard.isIndexing -->
           <!-- ko if: createWizard.isIndexing -->
             <span class="fa fa-spinner fa-spin"></span>
             <span class="fa fa-spinner fa-spin"></span>
           <!-- /ko -->
           <!-- /ko -->

+ 7 - 4
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -203,10 +203,13 @@ class Submission(object):
           notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=action.data['properties']['uuid']))
           notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=action.data['properties']['uuid']))
 
 
           self._create_file(deployment_dir, action.data['name'] + '.sql', notebook.get_str())
           self._create_file(deployment_dir, action.data['name'] + '.sql', notebook.get_str())
-        elif action.data['type'] == 'java-document':
-          from notebook.models import Notebook
-          notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=action.data['properties']['uuid']))
-          properties = notebook.get_data()['snippets'][0]['properties']
+        elif action.data['type'] == 'java-document' or action.data['type'] == 'java':
+          if action.data['type'] == 'java-document':
+            from notebook.models import Notebook
+            notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=action.data['properties']['uuid']))
+            properties = notebook.get_data()['snippets'][0]['properties']
+          else:
+            properties = action.data['properties']
 
 
           if properties.get('app_jar'):
           if properties.get('app_jar'):
             LOG.debug("Adding to oozie.libpath %s" % properties['app_jar'])
             LOG.debug("Adding to oozie.libpath %s" % properties['app_jar'])

+ 3 - 3
desktop/libs/notebook/src/notebook/api.py

@@ -101,11 +101,11 @@ def _execute_notebook(request, notebook, snippet):
   result = None
   result = None
   history = None
   history = None
 
 
-  is_query = notebook['type'].startswith('query-') or snippet['type'] == 'java'
+  historify = notebook['type'] != 'notebook' or snippet.get('wasBatchExecuted')
 
 
   try:
   try:
     try:
     try:
-      if is_query:
+      if historify:
         history = _historify(notebook, request.user)
         history = _historify(notebook, request.user)
         notebook = Notebook(document=history).get_data()
         notebook = Notebook(document=history).get_data()
 
 
@@ -115,7 +115,7 @@ def _execute_notebook(request, notebook, snippet):
       if response['handle'].get('sync'):
       if response['handle'].get('sync'):
         result = response['handle'].pop('result')
         result = response['handle'].pop('result')
     finally:
     finally:
-      if is_query:
+      if historify:
         _snippet = [s for s in notebook['snippets'] if s['id'] == snippet['id']][0]
         _snippet = [s for s in notebook['snippets'] if s['id'] == snippet['id']][0]
         if 'handle' in response: # No failure
         if 'handle' in response: # No failure
           _snippet['result']['handle'] = response['handle']
           _snippet['result']['handle'] = response['handle']

+ 5 - 4
desktop/libs/notebook/src/notebook/connectors/oozie_batch.py

@@ -60,11 +60,12 @@ class OozieApi(Api):
     if not notebook.get('uuid', ''):
     if not notebook.get('uuid', ''):
       raise PopupException(_('Notebook is missing a uuid, please save the notebook before executing as a batch job.'))
       raise PopupException(_('Notebook is missing a uuid, please save the notebook before executing as a batch job.'))
 
 
-    notebook_doc = Document2.objects.get_by_uuid(user=self.user, uuid=notebook['uuid'], perm_type='read')
-
-    if notebook_doc.type == 'notebook':
-      pass
+    if notebook['type'] == 'notebook':
+      # Convert notebook to workflow
+      workflow_doc = WorkflowBuilder().create_notebook_workflow(notebook=notebook, user=self.user, managed=True, name=_("Batch job for %s") % (notebook['name'] or notebook['type']))
+      workflow = Workflow(document=workflow_doc, user=self.user)
     else:
     else:
+      notebook_doc = Document2.objects.get_by_uuid(user=self.user, uuid=notebook['uuid'], perm_type='read')
       # Create a managed workflow from the notebook doc
       # Create a managed workflow from the notebook doc
       workflow_doc = WorkflowBuilder().create_workflow(document=notebook_doc, user=self.user, managed=True, name=_("Batch job for %s") % (notebook_doc.name or notebook_doc.type))
       workflow_doc = WorkflowBuilder().create_workflow(document=notebook_doc, user=self.user, managed=True, name=_("Batch job for %s") % (notebook_doc.name or notebook_doc.type))
       workflow = Workflow(document=workflow_doc, user=self.user)
       workflow = Workflow(document=workflow_doc, user=self.user)

+ 2 - 3
desktop/libs/notebook/src/notebook/models.py

@@ -145,11 +145,10 @@ def make_notebook2(name='Browse', description='', is_saved=False, snippets=None)
 
 
     _snippets.append(snippet)
     _snippets.append(snippet)
 
 
-  print _snippets
-
   data = {
   data = {
     'name': name,
     'name': name,
     'uuid': str(uuid.uuid4()),
     'uuid': str(uuid.uuid4()),
+    'type': 'notebook',
     'description': description,
     'description': description,
     'sessions': [
     'sessions': [
       {
       {
@@ -169,7 +168,7 @@ def make_notebook2(name='Browse', description='', is_saved=False, snippets=None)
          'statement_raw': _snippet.get('statement', ''),
          'statement_raw': _snippet.get('statement', ''),
          'statement': _snippet.get('statement', ''),
          'statement': _snippet.get('statement', ''),
          'type': _snippet.get('type'),
          'type': _snippet.get('type'),
-         'properties': _snippet.properties,
+         'properties': _snippet['properties'],
          'name': name,
          'name': name,
          'database': _snippet.get('database'),
          'database': _snippet.get('database'),
          'result': {}
          'result': {}