Pārlūkot izejas kodu

HUE-4490 [oozie] Generate workflow from Java snippet

Romain Rigaux 9 gadi atpakaļ
vecāks
revīzija
383d91f

+ 9 - 0
desktop/libs/indexer/src/indexer/smart_indexer.py

@@ -84,6 +84,15 @@ class Indexer(object):
   def run_morphline(self, collection_name, morphline, input_path):
     workspace_path = self._upload_workspace(morphline)
 
+    notebook_doc = Document2.objects.get_by_uuid(user=self.user, uuid=notebook['uuid'], perm_type='read')
+
+    # 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)
+    workflow = Workflow(document=workflow_doc, user=self.user)
+
+    # Submit workflow
+    job_id = _submit_workflow(user=self.user, fs=self.fs, jt=self.jt, workflow=workflow, mapping=None)
+
     job_id = self._schedule_oozie_job(workspace_path, collection_name, input_path)
     return job_id
 

+ 15 - 12
desktop/libs/indexer/src/indexer/tests_indexer.py

@@ -19,19 +19,20 @@ import logging
 from nose.tools import assert_equal, assert_true
 from nose.plugins.skip import SkipTest
 
-from hadoop import cluster
-from hadoop.pseudo_hdfs4 import is_live_cluster
+from hadoop.pseudo_hdfs4 import is_live_cluster, shared_cluster
 
-from indexer.smart_indexer import Indexer
+from indexer.file_format import ApacheCombinedFormat, RubyLogFormat, HueLogFormat
+from indexer.fields import Field
 from indexer.controller import CollectionManagerController
 from indexer.operations import get_operator
-from indexer.file_format import ApacheCombinedFormat, RubyLogFormat, HueLogFormat
-from indexer.fields import Field, get_field_type
+from indexer.smart_indexer import Indexer
+
 
 LOG = logging.getLogger(__name__)
 
+
 def _test_fixed_type_format_generate_morphline(format_):
-  indexer = Indexer("test", None)
+  indexer = Indexer("test")
   format_instance = format_()
 
   morphline = indexer.generate_morphline_config("test_collection", {
@@ -45,7 +46,7 @@ def _test_generate_field_operation_morphline(operation_format):
   fields = IndexerTest.simpleCSVFields[:]
   fields[0]['operations'].append(operation_format)
 
-  indexer = Indexer("test", None)
+  indexer = Indexer("test")
   morphline =indexer.generate_morphline_config("test_collection", {
       "columns": fields,
       "format": IndexerTest.simpleCSVFormat
@@ -53,6 +54,7 @@ def _test_generate_field_operation_morphline(operation_format):
 
   assert_true(isinstance(morphline, basestring))
 
+
 class IndexerTest():
   simpleCSVString = """id,Rating,Location,Name,Time
 1,5,San Francisco,Good Restaurant,8:30pm
@@ -106,7 +108,7 @@ class IndexerTest():
 
   def test_guess_csv_format(self):
     stream = StringIO.StringIO(IndexerTest.simpleCSVString)
-    indexer = Indexer("test", None)
+    indexer = Indexer("test")
 
     guessed_format = indexer.guess_format({'file': {"stream": stream, "name": "test.csv"}})
 
@@ -124,7 +126,7 @@ class IndexerTest():
         assert_equal(expected[key], actual[key])
 
   def test_guess_format_invalid_csv_format(self):
-    indexer = Indexer("test", None)
+    indexer = Indexer("test")
     stream = StringIO.StringIO(IndexerTest.simpleCSVString)
 
     guessed_format = indexer.guess_format({'file': {"stream": stream, "name": "test.csv"}})
@@ -151,7 +153,7 @@ class IndexerTest():
     assert_equal(fields, [])
 
   def test_generate_csv_morphline(self):
-    indexer = Indexer("test", None)
+    indexer = Indexer("test")
     morphline =indexer.generate_morphline_config("test_collection", {
         "columns": self.simpleCSVFields,
         "format": self.simpleCSVFormat
@@ -234,9 +236,10 @@ class IndexerTest():
     if not is_live_cluster():
       raise SkipTest()
 
-    fs = cluster.get_hdfs()
+    cluster = shared_cluster()
+    fs = cluster.fs
     collection_name = "test_collection"
-    indexer = Indexer("test", fs)
+    indexer = Indexer("test", fs=fs, jt=cluster.jt)
     input_loc = "/tmp/test.csv"
 
     # upload the test file to hdfs

+ 1 - 0
desktop/libs/liboozie/src/liboozie/oozie_api.py

@@ -40,6 +40,7 @@ def get_oozie(user, api_version=API_VERSION):
   oozie_url = OOZIE_URL.get()
   secure = SECURITY_ENABLED.get()
   ssl_cert_ca_verify = SSL_CERT_CA_VERIFY.get()
+
   return OozieApi(oozie_url, user, security_enabled=secure, api_version=api_version, ssl_cert_ca_verify=ssl_cert_ca_verify)
 
 

+ 16 - 11
desktop/libs/notebook/src/notebook/api.py

@@ -281,29 +281,23 @@ def get_logs(request):
 
   return JsonResponse(response)
 
-
-@require_POST
-@check_document_modify_permission()
-def save_notebook(request):
-  response = {'status': -1}
-
-  notebook = json.loads(request.POST.get('notebook', '{}'))
+def _save_notebook(notebook, user):
   notebook_type = notebook.get('type', 'notebook')
   save_as = True
 
   if notebook.get('parentSavedQueryUuid'): # We save into the original saved query, not into the query history
-    notebook_doc = Document2.objects.get_by_uuid(user=request.user, uuid=notebook['parentSavedQueryUuid'])
+    notebook_doc = Document2.objects.get_by_uuid(user=user, uuid=notebook['parentSavedQueryUuid'])
   elif notebook.get('id'):
     notebook_doc = Document2.objects.get(id=notebook['id'])
   else:
-    notebook_doc = Document2.objects.create(name=notebook['name'], uuid=notebook['uuid'], type=notebook_type, owner=request.user)
+    notebook_doc = Document2.objects.create(name=notebook['name'], uuid=notebook['uuid'], type=notebook_type, owner=user)
     Document.objects.link(notebook_doc, owner=notebook_doc.owner, name=notebook_doc.name, description=notebook_doc.description, extra=notebook_type)
     save_as = False
 
     if notebook.get('directoryUuid'):
-      notebook_doc.parent_directory = Document2.objects.get_by_uuid(user=request.user, uuid=notebook.get('directoryUuid'), perm_type='write')
+      notebook_doc.parent_directory = Document2.objects.get_by_uuid(user=user, uuid=notebook.get('directoryUuid'), perm_type='write')
     else:
-      notebook_doc.parent_directory = Document2.objects.get_home_directory(request.user)
+      notebook_doc.parent_directory = Document2.objects.get_home_directory(user)
 
   notebook['isSaved'] = True
   notebook['isHistory'] = False
@@ -326,6 +320,17 @@ def save_notebook(request):
   notebook_doc.description = notebook_doc1.description = notebook['description']
   notebook_doc.save()
   notebook_doc1.save()
+  
+  return notebook_doc, save_as
+
+@require_POST
+@check_document_modify_permission()
+def save_notebook(request):
+  response = {'status': -1}
+
+  notebook = json.loads(request.POST.get('notebook', '{}'))
+
+  notebook_doc, save_as = _save_notebook(notebook, request.user)
 
   response['status'] = 0
   response['save_as'] = save_as

+ 42 - 10
desktop/libs/notebook/src/notebook/models.py

@@ -57,29 +57,56 @@ def make_notebook(name='Browse', description='', editor_type='hive', statement='
   from notebook.connectors.hiveserver2 import HS2Api
 
   editor = Notebook()
-
-  properties = HS2Api.get_properties(editor_type)
+  extra_snippet_properties = {}
 
   if editor_type == 'hive':
+    sessions_properties = HS2Api.get_properties(editor_type)
     if files is not None:
-      _update_property_value(properties, 'files', files)
+      _update_property_value(sessions_properties, 'files', files)
 
     if functions is not None:
-      _update_property_value(properties, 'functions', functions)
+      _update_property_value(sessions_properties, 'functions', functions)
 
     if settings is not None:
-      _update_property_value(properties, 'settings', settings)
+      _update_property_value(sessions_properties, 'settings', settings)
   elif editor_type == 'impala':
+    sessions_properties = HS2Api.get_properties(editor_type)
     if settings is not None:
-      _update_property_value(properties, 'files', files)
-
-  editor.data = json.dumps({
+      _update_property_value(sessions_properties, 'files', files)
+  elif editor_type == 'java':
+    sessions_properties = [] # Java options
+    extra_snippet_properties =  {
+      u'files': [
+          {u'path': u'/user/romain/tmp/log4j.properties', u'type': u'file'},
+          {u'path': u'/user/hue/oozie/workspaces/hue-oozie-1469837046.14/morphline.conf', u'type': u'file'}
+      ],
+      u'class': u'org.apache.solr.hadoop.MapReduceIndexerTool',
+      u'app_jar': u'/tmp/smart_indexer_lib',
+      u'arguments': [
+          u'--morphline-file',
+          u'morphline.conf',
+          u'--output-dir',
+          u'${nameNode}/tmp/editor',
+          u'--log4j',
+          u'log4j.properties',
+          u'--go-live',
+          u'--zk-host',
+          u'localhost:2181/solr',
+          u'--collection',
+          u'earthquakes',
+          u'${nameNode}/user/romain/2.5_month.csv',
+      ],
+      u'archives': [],
+    }
+
+  data = {
     'name': name,
+    'uuid': str(uuid.uuid4()),
     'description': description,
     'sessions': [
       {
          'type': editor_type,
-         'properties': properties,
+         'properties': sessions_properties,
          'id': None
       }
     ],
@@ -104,7 +131,12 @@ def make_notebook(name='Browse', description='', editor_type='hive', statement='
          'result': {}
       }
     ]
-  })
+  }
+
+  if extra_snippet_properties:
+    data['snippets'][0]['properties'].update(extra_snippet_properties)
+
+  editor.data = json.dumps(data)
   
   return editor