Browse Source

HUE-8208 [importer] Integrate Kafka input with Envelope Spark lib

Romain Rigaux 7 years ago
parent
commit
74a29db3f6

+ 7 - 6
apps/oozie/src/oozie/models2.py

@@ -3741,6 +3741,8 @@ class WorkflowBuilder():
         node = self.get_hive_snippet_node(snippet, user)
       elif snippet['type'] == 'impala':
         node = self.get_impala_snippet_node(snippet, user)
+      elif snippet['type'] == 'spark':
+        node = self.get_spark_snippet_node(snippet, user)
       elif snippet['type'] == 'shell':
         node = self.get_shell_snippet_node(snippet)
       else:
@@ -3879,7 +3881,7 @@ class WorkflowBuilder():
     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"type": u"spark2-document-widget" if is_document_node else u"spark-widget",
         u'properties': {
             u'files': [],
             u'job_xml': u'',
@@ -3911,15 +3913,14 @@ class WorkflowBuilder():
         u'actionParameters': [],
     }
 
-  def get_spark_snippet_node(self, snippet):
-    credentials = []
-
+  def get_spark_snippet_node(self, snippet, user):
     node_id = snippet.get('id', str(uuid.uuid4()))
 
-    node = self._get_java_node(node_id, credentials)
+    node = self._get_spark_node(node_id, user)
     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']['files'] = [{'value': f['path']} for f in snippet['properties']['files']]
+    node['properties']['spark_opts'] = snippet['properties']['spark_opts']
     node['properties']['spark_arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
 
     return node

+ 3 - 0
desktop/core/src/desktop/templates/ko_components/ko_history_panel.mako

@@ -177,6 +177,9 @@ from desktop.views import _ko
             sqoop1: {
               sqlDialect: false
             },
+            spark: {
+              sqlDialect: false
+            },
           }
         });
         self.editorViewModel.editorMode(true);

File diff suppressed because it is too large
+ 3 - 0
desktop/libs/indexer/src/indexer/api3.py


+ 7 - 0
desktop/libs/indexer/src/indexer/conf.py

@@ -103,6 +103,13 @@ ENABLE_SQOOP = Config(
   default=False
 )
 
+ENABLE_KAFKA = Config(
+  key="enable_kafka",
+  help=_t("Flag to turn on Kafka imports."),
+  type=bool,
+  default=False
+)
+
 # Unused
 BATCH_INDEXER_PATH = Config(
   key="batch_indexer_path",

+ 168 - 0
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -0,0 +1,168 @@
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.import logging
+
+import logging
+import os
+
+from django.contrib.auth.models import User
+from django.core.urlresolvers import reverse
+from django.utils.translation import ugettext as _
+
+from notebook.models import make_notebook
+
+
+LOG = logging.getLogger(__name__)
+
+
+class EnvelopeIndexer(object):
+
+  def __init__(self, username, fs=None, jt=None, solr_client=None):
+    self.fs = fs
+    self.jt = jt
+    self.username = username
+
+
+  def _upload_workspace(self, morphline):
+    from oozie.models2 import Job
+
+    hdfs_workspace_path = Job.get_workspace(self.username)
+    hdfs_morphline_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)
+
+    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)
+
+    task = make_notebook(
+      name=_('Indexing into %s') % collection_name,
+      editor_type='notebook',
+      #on_success_url=reverse('search:browse', kwargs={'name': collection_name}),
+      #pub_sub_url='assist.collections.refresh',
+      is_task=True,
+      is_notebook=True,
+      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'}
+      ]
+    )
+
+    return task.execute(request, batch=True)
+
+
+  def generate_config(self):
+    properties = {
+      "brokers": "self-service-analytics-1.gce.cloudera.com:9092,self-service-analytics-2.gce.cloudera.com:9092,self-service-analytics-3.gce.cloudera.com:9092",
+      "kudu_master": "self-service-analytics-1.gce.cloudera.com:7051",
+      "output_table": "impala::default.traffic_conditions"
+    }
+
+    return """
+application {
+    name = Traffic analysis
+    batch.milliseconds = 5000
+    executors = 1
+    executor.cores = 1
+    executor.memory = 1G
+}
+
+steps {
+    traffic {
+        input {
+            type = kafka
+            brokers = "%(brokers)s"
+            topics = traffic
+            encoding = string
+            translator {
+                type = delimited
+                delimiter = ","
+                field.names = [measurement_time,number_of_vehicles]
+                field.types = [long,int]
+            }
+            window {
+                enabled = true
+                milliseconds = 60000
+            }
+        }
+    }
+
+    trafficwindow {
+        dependencies = [traffic]
+        deriver {
+            type = sql
+            query.literal = \"""
+                SELECT UNIX_TIMESTAMP() * 1000 as_of_time, ROUND(AVG(number_of_vehicles), 2) avg_num_veh,
+                MIN(number_of_vehicles) min_num_veh, MAX(number_of_vehicles) max_num_veh,
+                MIN(measurement_time) first_meas_time, MAX(measurement_time) last_meas_time FROM traffic\"""
+        }
+        planner {
+            type = upsert
+        }
+        output {
+            type = kudu
+            connection = "%(kudu_master)s"
+            table.name = "%(output_table)s"
+        }
+    }
+}
+
+""" % properties
+
+
+"""application {
+  name = Filesystem Example
+  executors = 1
+}
+steps {
+  fsInput {
+    input {
+      type = filesystem
+      // Be sure to load this file into HDFS first!
+      path = example-input.json
+      format = json
+    }
+  }
+  fsProcess {
+    dependencies = [fsInput]
+    deriver {
+      type = sql
+      query.literal = "SELECT foo FROM fsInput"
+    }
+    planner = {
+      type = overwrite
+    }
+    output = {
+      type = filesystem
+      // The output directory
+      path = example-output
+      format = parquet
+    }
+  }
+}
+"""

+ 69 - 10
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -20,7 +20,7 @@
   from desktop import conf
   from desktop.views import commonheader, commonfooter, commonshare, commonimportexport, _ko
 
-  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER
+  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, ENABLE_KAFKA, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER
   from notebook.conf import ENABLE_SQL_INDEXER
 %>
 
@@ -248,14 +248,14 @@ ${ assist.assistPanel() }
             <!-- /ko -->
 
             <!-- ko if: createWizard.source.rdbmsMode() == 'customRdbms' -->
-              <div class="control-group input-append">
+              <div class="control-group">
                 <label for="rdbmsHostname" class="control-label"><div>${ _('Hostname') }</div>
-                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsHostname" placeholder="${ _('Enter host/ip here eg. mysql.domain.com or 123.123.123.123') }">
+                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsHostname" placeholder="${ _('Enter host/ip here e.g. mysql.domain.com or 123.123.123.123') }">
                 </label>
               </div>
 
               <!-- ko if: createWizard.source.rdbmsType() == 'jdbc' -->
-              <div class="control-group input-append">
+              <div class="control-group">
                 <label for="rdbmsJdbcDriver" class="control-label"><div>${ _('JDBC Driver') }</div>
                   <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsJdbcDriver">
                 </label>
@@ -265,7 +265,7 @@ ${ assist.assistPanel() }
               <!-- ko if: createWizard.source.rdbmsType() != 'jdbc' -->
               <div class="control-group">
                 <label for="rdbmsPort" class="control-label"><div>${ _('Port') }</div>
-                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsPort" placeholder="${ _('Enter port number here eg. 3306') }">
+                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.rdbmsPort" placeholder="${ _('Enter port number here e.g. 3306') }">
                 </label>
               </div>
               <!-- /ko -->
@@ -322,6 +322,22 @@ ${ assist.assistPanel() }
 
           <!-- /ko -->
 
+          <!-- ko if: createWizard.source.inputFormat() == 'kafka' -->
+            ## Service
+
+            <div class="control-group">
+              <label for="rdbmsHostname" class="control-label"><div>${ _('Brokers') }</div>
+                <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaBrokers" placeholder="${ _('Enter a csv list of brokers, e.g.brokers1:9092,brokers2:9092') }">
+              </label>
+            </div>
+
+            <div class="control-group">
+              <label for="rdbmsHostname" class="control-label"><div>${ _('Topic') }</div>
+                <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaTopics" placeholder="${ _('The list of topics to consume, e.g. orders,returns') }">
+              </label>
+            </div>
+          <!-- /ko -->
+
           <div class="control-group" data-bind="visible: createWizard.source.inputFormat() == 'table'">
             <label for="path" class="control-label"><div>${ _('Table') }</div>
               <input type="text" class="input-xlarge" data-bind="value: createWizard.source.table, hivechooser: createWizard.source.table, skipColumns: true, apiHelperUser: '${ user }', apiHelperType: createWizard.source.apiHelperType, mainScrollable: $(MAIN_SCROLLABLE)" placeholder="${ _('Table name or <database>.<table>') }">
@@ -415,13 +431,13 @@ ${ assist.assistPanel() }
           <div class="control-group">
             <label for="collectionName" class="control-label "><div>${ _('Name') }</div></label>
             <!-- ko if: outputFormat() == 'file' -->
-              <input type="text" class="form-control name input-xlarge" id="collectionName" data-bind="value: name, filechooser: name, filechooserOptions: { linkMarkup: true, skipInitialPathIfEmpty: true, openOnFocus: true, selectFolder: true, displayOnlyFolders: true, uploadFile: false}" placeholder="${ _('Name') }" title="${ _('Directory must not exist in the path') }">
+              <input type="text" class="form-control name input-xxlarge" id="collectionName" data-bind="value: name, filechooser: name, filechooserOptions: { linkMarkup: true, skipInitialPathIfEmpty: true, openOnFocus: true, selectFolder: true, displayOnlyFolders: true, uploadFile: false}" placeholder="${ _('Name') }" title="${ _('Directory must not exist in the path') }">
             <!-- /ko -->
             <!-- ko if: outputFormat() == 'index' -->
               <input type="text" class="form-control input-xlarge" id="collectionName" data-bind="value: name, valueUpdate: 'afterkeydown'" placeholder="${ _('Name') }">
             <!-- /ko -->
             <!-- ko if: ['table', 'database'].indexOf(outputFormat()) != -1 -->
-              <input type="text" data-bind="value: name, hivechooser: name, skipColumns: true, skipTables: outputFormat() == 'database', valueUpdate: 'afterkeydown', apiHelperUser: '${ user }', apiHelperType: apiHelperType, mainScrollable: $(MAIN_SCROLLABLE), attr: { 'placeholder': outputFormat() == 'table' ? '${  _ko('Table name or <database>.<table>') }' : '${  _ko('Database name') }' }" pattern="^([a-zA-Z0-9_]+\.)?[a-zA-Z0-9_]*$" title="${ _('Only alphanumeric and underscore characters') }">
+              <input type="text" class="input-xlarge" data-bind="value: name, hivechooser: name, skipColumns: true, skipTables: outputFormat() == 'database', valueUpdate: 'afterkeydown', apiHelperUser: '${ user }', apiHelperType: apiHelperType, mainScrollable: $(MAIN_SCROLLABLE), attr: { 'placeholder': outputFormat() == 'table' ? '${  _ko('Table name or <database>.<table>') }' : '${  _ko('Database name') }' }" pattern="^([a-zA-Z0-9_]+\.)?[a-zA-Z0-9_]*$" title="${ _('Only alphanumeric and underscore characters') }">
             <!-- /ko -->
             <span class="help-inline muted" data-bind="visible: !isTargetExisting() && isTargetChecking()">
               <i class="fa fa-spinner fa-spin"></i>
@@ -717,7 +733,34 @@ ${ assist.assistPanel() }
         </div>
         <!-- /ko -->
 
-        <!-- ko if: ['table', 'index', 'file', 'hbase'].indexOf(outputFormat()) != -1 -->
+        <!-- ko if: $root.createWizard.source.inputFormat() == 'kafka' -->
+        <div class="card step">
+          <h4>${_('Properties')}</h4>
+
+          <div class="card-body">
+            <label class="control-label"><div>${ _('Libs') }</div>
+              <div class="inline-table">
+                <ul data-bind="sortable: { data: sqoopJobLibPaths, options: { axis: 'y', containment: 'parent', handle: '.move-widget' }}, visible: sqoopJobLibPaths().length" class="unstyled">
+                  <li>
+                    <div class="input-append" style="margin-bottom: 4px">
+                      <input type="text" class="filechooser-input input-xxlarge" data-bind="value: path, valueUpdate:'afterkeydown', filechooser: { value: path, isAddon: true }, filechooserOptions: { skipInitialPathIfEmpty: true }" placeholder="${ _('Path to the file, e.g. hdfs://localhost:8020/user/hue/file.hue') }"/>
+                      <span class="add-on move-widget muted" data-bind="visible: $parent.sqoopJobLibPaths().length > 1"><i class="fa fa-arrows"></i></span>
+                      <a class="add-on muted" href="javascript: void(0);" data-bind="click: function(){ $parent.removeSqoopJobLibPath($data); }"><i class="fa fa-minus"></i></a>
+                    </div>
+                  </li>
+                </ul>
+                <div class="config-property-add-value" style="margin-top: 5px;">
+                  <a class="inactive-action pointer" style="padding: 3px 10px 3px 3px;;" data-bind="click: addSqoopJobLibPath">
+                    <i class="fa fa-plus"></i>
+                  </a>
+                </div>
+              </div>
+            </label>
+          </div>
+        </div>
+        <!-- /ko -->
+
+        <!-- ko if: ['table', 'index', 'hbase'].indexOf(outputFormat()) != -1 -->
           <div class="card step">
             <h4 class="show-edit-on-hover">${_('Fields')} <!-- ko if: $root.createWizard.isGuessingFieldTypes --><i class="fa fa-spinner fa-spin"></i><!-- /ko --> <a class="inactive-action pointer" data-bind="visible: columns().length > 0" href="#fieldsBulkEditor" data-toggle="modal"><i class="fa fa-edit"></i></a></h4>
             <div class="card-body no-margin-top columns-form">
@@ -1250,6 +1293,9 @@ ${ assist.assistPanel() }
           % if ENABLE_SQOOP.get():
           {'value': 'rdbms', 'name': 'External Database'},
           % endif
+          % if ENABLE_KAFKA.get():
+          {'value': 'kafka', 'name': 'Kafka Stream'},
+          % endif
           % if ENABLE_SQL_INDEXER.get():
           {'value': 'query', 'name': 'SQL Query'},
           ##{'value': 'table', 'name': 'Table'},
@@ -1443,6 +1489,15 @@ ${ assist.assistPanel() }
       });
       self.draggedQuery = ko.observable();
 
+      // Kafka
+      self.kafkaBrokers = ko.observable('brokers1:9092,brokers2:9092');
+      self.kafkaTopics = ko.observable('');
+      self.kafkaTopics.subscribe(function(newValue) {
+        if (newValue) {
+          viewModel.createWizard.guessFieldTypes();
+        }
+      });
+
       self.format = ko.observable();
       self.format.subscribe(function(newVal) {
         if (typeof newVal.hasHeader !== 'undefined') {
@@ -1472,6 +1527,8 @@ ${ assist.assistPanel() }
           return self.query();
         } else if (self.inputFormat() == 'manual') {
           return true;
+        } else if (self.inputFormat() == 'kafka') {
+          return self.kafkaBrokers().length > 0 && self.kafkaTopics().length > 0;
         } else if (self.inputFormat() == 'rdbms') {
           return self.rdbmsDatabaseName().length > 0 && (self.rdbmsTableName().length > 0 || self.rdbmsAllTablesSelected());
         }
@@ -1570,8 +1627,10 @@ ${ assist.assistPanel() }
           {'name': 'Search index', 'value': 'index'},
           % endif
           {'name': 'Database', 'value': 'database'},
-          % if ENABLE_SQOOP.get():
+          % if ENABLE_SQOOP.get() or ENABLE_KAFKA.get():
           {'name': 'File', 'value': 'file'},
+          % endif
+          % if ENABLE_SQOOP.get():
           {'name': 'HBase Table', 'value': 'hbase'},
           % endif
       ]);
@@ -1580,7 +1639,7 @@ ${ assist.assistPanel() }
           if (format.value == 'database' && wizard.source.inputFormat() != 'manual') {
             return false;
           }
-          if (format.value == 'file' && ['manual', 'rdbms'].indexOf(wizard.source.inputFormat()) == -1) {
+          if (format.value == 'file' && ['manual', 'rdbms', 'kafka'].indexOf(wizard.source.inputFormat()) == -1) {
             return false;
           }
           else if (format.value == 'index' && ['file', 'query', 'table'].indexOf(wizard.source.inputFormat()) == -1) {

+ 19 - 0
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -154,6 +154,25 @@ class Notebook(object):
 
     self.data = json.dumps(_data)
 
+  def add_spark_snippet(self, clazz, jars, arguments, files):
+    _data = json.loads(self.data)
+
+    _data['snippets'].append(self._make_snippet({
+        u'type': u'spark',
+        u'status': u'running',
+        u'properties':  {
+          u'files': files,
+          u'class': clazz,
+          u'app_jar': jars,
+          u'arguments': arguments,
+          u'archives': [],
+          u'spark_opts': ''
+        }
+    }))
+    self._add_session(_data, 'spark')
+
+    self.data = json.dumps(_data)
+
   def add_shell_snippet(self, shell_command, arguments, archives, files, env_var, last_executed):
     _data = json.loads(self.data)
 

Some files were not shown because too many files changed in this diff