Browse Source

HUE-8208 [importer] Alternatively submit envelop jobs with a Oozie shell action

Romain Rigaux 7 years ago
parent
commit
4b7b8c93e9

+ 5 - 4
desktop/libs/indexer/src/indexer/api3.py

@@ -449,7 +449,7 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
   collection_name = destination['name']
   indexer = EnvelopeIndexer(request.user, request.fs)
 
-  lib_path = lib_path or '/tmp/envelope-0.5.0.jar'
+  lib_path = '/tmp/envelope-0.5.0.jar'
   input_path = None
 
   if file_format['inputFormat'] == 'table':
@@ -490,7 +490,8 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
       else:
         sql = SQLIndexer(user=request.user, fs=request.fs).create_table_from_a_file(file_format, destination).get_str()
         print sql
-      if file_format['inputFormat'] == 'stream':
+      if destination['tableFormat'] == 'kudu':
+        manager = ManagerApi()
         properties["output_table"] = "impala::%s" % collection_name
         properties["kudu_master"] = manager.get_kudu_master()
       else:
@@ -514,9 +515,9 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
   properties["ouputFormat"] = destination['ouputFormat']
   properties["streamSelection"] = file_format["streamSelection"]
 
-  morphline = indexer.generate_config(properties)
+  envelope = indexer.generate_config(properties)
 
-  return indexer.run(request, collection_name, morphline, input_path, start_time=start_time, lib_path=lib_path)
+  return indexer.run(request, collection_name, envelope, input_path, start_time=start_time, lib_path=lib_path)
 
 
 def _create_solr_collection(user, fs, client, destination, index_name, kwargs):

+ 33 - 16
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -23,6 +23,7 @@ from django.utils.translation import ugettext as _
 
 from notebook.models import make_notebook
 from desktop.lib.exceptions_renderable import PopupException
+from desktop.conf import DISABLE_HUE_3
 
 
 LOG = logging.getLogger(__name__)
@@ -36,22 +37,22 @@ class EnvelopeIndexer(object):
     self.username = username
 
 
-  def _upload_workspace(self, morphline):
+  def _upload_workspace(self, envelope):
     from oozie.models2 import Job
 
     hdfs_workspace_path = Job.get_workspace(self.username)
-    hdfs_morphline_path = os.path.join(hdfs_workspace_path, "envelope.conf")
+    hdfs_envelope_path = os.path.join(hdfs_workspace_path, "envelope.conf")
 
     # Create workspace on hdfs
     self.fs.do_as_user(self.username, self.fs.mkdir, hdfs_workspace_path)
 
-    self.fs.do_as_user(self.username, self.fs.create, hdfs_morphline_path, data=morphline)
+    self.fs.do_as_user(self.username, self.fs.create, hdfs_envelope_path, data=envelope)
 
     return hdfs_workspace_path
 
 
-  def run(self, request, collection_name, morphline, input_path, start_time=None, lib_path=None):
-    workspace_path = self._upload_workspace(morphline)
+  def run(self, request, collection_name, envelope, input_path, start_time=None, lib_path=None):
+    workspace_path = self._upload_workspace(envelope)
 
     task = make_notebook(
       name=_('Indexing into %s') % collection_name,
@@ -63,16 +64,32 @@ class EnvelopeIndexer(object):
       last_executed=start_time
     )
 
-    task.add_spark_snippet(
-      clazz=None,
-      jars=lib_path,
-      arguments=[
-          u'envelope.conf'
-      ],
-      files=[
-          {u'path': u'%s/envelope.conf' % workspace_path, u'type': u'file'}
-      ]
-    )
+    if not DISABLE_HUE_3.get(): # CDH5
+      shell_command_name = "pipeline.sh"
+      shell_command = """#!/bin/bash
+
+SPARK_KAFKA_VERSION=0.10 spark2-submit envelope.jar envelope.conf"""
+      hdfs_shell_cmd_path = os.path.join(workspace_path, shell_command_name)
+      self.fs.do_as_user(self.username, self.fs.create, hdfs_shell_cmd_path, data=shell_command)
+      task.add_shell_snippet(
+        shell_command=shell_command_name,
+        files=[
+            {u'value': u'%s/envelope.conf' % workspace_path},
+            {u'value': hdfs_shell_cmd_path},
+            {u'value': lib_path, }
+        ]
+      )
+    else:
+      task.add_spark_snippet(
+        clazz=None,
+        jars=lib_path,
+        arguments=[
+            u'envelope.conf'
+        ],
+        files=[
+            {u'path': u'%s/envelope.conf' % workspace_path, u'type': u'file'}
+        ]
+      )
 
     return task.execute(request, batch=True)
 
@@ -167,7 +184,7 @@ class EnvelopeIndexer(object):
         }""" % properties
     else:
       raise PopupException(_('Output format not recognized: %(ouputFormat)s') % properties)
-      
+
     return """
 application {
     name = %(app_name)s

+ 6 - 1
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -1634,13 +1634,18 @@ ${ assist.assistPanel() }
       self.streamEndpointUrl = ko.observable('https://login.salesforce.com/services/Soap/u/42.0');
       self.streamObjects = ko.observableArray();
       self.streamObject = ko.observable('');
+      self.streamObject.subscribe(function(newValue) {
+        if (newValue) {
+          wizard.guessFieldTypes();
+        }
+      });
       self.hasStreamSelected = ko.pureComputed(function() {
         return (self.streamSelection() == 'kafka' && self.kafkaSelectedTopics()) ||
            (self.streamSelection() == 'sfdc' && self.streamObject())
       });
       self.hasStreamSelected.subscribe(function(newValue) {
         if (newValue) {
-          wizard.guessFormat();
+          wizard.guessFormat(); 
           if (newValue == 'kafka') {
             wizard.destination.tableFormat('kudu');
           }

+ 11 - 2
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -173,9 +173,18 @@ class Notebook(object):
 
     self.data = json.dumps(_data)
 
-  def add_shell_snippet(self, shell_command, arguments, archives, files, env_var, last_executed):
+  def add_shell_snippet(self, shell_command, arguments=None, archives=None, files=None, env_var=None, last_executed=None):
     _data = json.loads(self.data)
 
+    if arguments is None:
+      arguments = []
+    if archives is None:
+      archives = []
+    if files is None:
+      files = []
+    if env_var is None:
+      env_var = []
+
     _data['snippets'].append(self._make_snippet({
         u'type': u'shell',
         u'status': u'running',
@@ -252,7 +261,7 @@ def get_api(request, snippet):
         'interface': 'kafka',
         'options': {},
         'is_sql': False
-      }]      
+      }]
     elif snippet['type'] == 'solr':
       interpreter = [{
         'name': 'solr',