Browse Source

HUE-8208 [importer] Polish the selection of an input stream

Romain Rigaux 7 years ago
parent
commit
84b30333c2

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


+ 21 - 18
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -78,24 +78,24 @@ class EnvelopeIndexer(object):
 
 
   def generate_config(self, properties):
-    if properties['inputFormat'] == 'kafka':
-#               translator {
-#                   type = delimited
-#                   delimiter = ","
-#                   field.names = [measurement_time,number_of_vehicles]
-#                   field.types = [long,int]
-#               }
-      input = """type = kafka
-              brokers = "%(brokers)s"
-              topics = %(topics)s
-              encoding = string
-              window {
-                  enabled = true
-                  milliseconds = 60000
-              }
-      """ % properties
-    elif properties['inputFormat'] == 'stream':
-      if properties['streamSelection'] == 'sfdc':
+    if properties['inputFormat'] == 'stream':
+      if properties['streamSelection'] == 'kafka':
+        input = """type = kafka
+                brokers = "%(brokers)s"
+                topics = %(topics)s
+                encoding = string
+                translator {
+                    type = %(kafkaFieldType)s
+                    delimiter = "%(kafkaFieldDelimiter)s"
+                    field.names = [%(kafkaFieldNames)s]
+                    field.types = [%(kafkaFieldTypes)s]
+                }
+                window {
+                    enabled = true
+                    milliseconds = 60000
+                }
+        """ % properties
+      elif properties['streamSelection'] == 'sfdc':
         input = """type = sfdc
         mode = fetch-all
         sobject = %(streamObject)s
@@ -108,6 +108,8 @@ class EnvelopeIndexer(object):
           }
         }
   """
+      else:
+        raise PopupException(_('Stream format of %(inputFormat)s not recognized: %(streamSelection)s') % properties)
     elif properties['inputFormat'] == 'file':
       input = """type = filesystem
       path = %(path)s
@@ -116,6 +118,7 @@ class EnvelopeIndexer(object):
     else:
       raise PopupException(_('Input format not recognized: %(inputFormat)s') % properties)
 
+
     if properties['ouputFormat'] == 'file':
       output = """dependencies = [inputdata]
     planner = {

+ 51 - 18
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -340,15 +340,41 @@ ${ assist.assistPanel() }
 
                 <br/>
 
-##                <span data-bind="visible: createWizard.source.kafkaSelectedTopics">
-##          <label class="control-label"><div>${ _('Field names') }</div>
-##                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaFieldNames" placeholder="${ _('The list of fields to consume, e.g. orders,returns') }">
-##                </label>
-
-##                <label class="control-label"><div>${ _('Field types') }</div>
-##                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaFieldTypes" placeholder="${ _('The list of topics to consume, e.g. orders,returns') }">
-##                </label>
-##              </span>
+                <div class="control-group" data-bind="visible: createWizard.source.kafkaSelectedTopics">
+                  <label class="control-label"><div>${ _('Schema') }</div>
+                    <label class="checkbox inline-block">
+                      <input type="radio" name="kafkaSchemaManual" value="manual" data-bind="checked: createWizard.source.kafkaSchemaManual" /> ${_('Manual')}
+                    </label>
+                    <label class="checkbox inline-block">
+                      <input type="radio" name="kafkaSchemaManual" value="detect" data-bind="checked: createWizard.source.kafkaSchemaManual" /> ${_('Guess')}
+                    </label>
+                  </label>
+
+                  <label class="control-label" data-bind="visible: createWizard.source.kafkaSchemaManual() == 'manual'">
+                  ##<label class="control-label"><div>${ _('Encoding') }</div>
+                  ##  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaFieldType">
+                  ##</label>
+                  <label class="control-label"><div>${ _('Type') }</div>
+                    <select class="input-small" data-bind="options: ['delimited', 'bitarray'], value: createWizard.source.kafkaFieldType"></select>
+                  </label>
+                  <label class="control-label"><div>${ _('Delimiter') }</div>
+                    <input type="text" class="input-small" data-bind="value: createWizard.source.kafkaFieldDelimiter">
+                  </label>
+
+                  <label class="control-label"><div>${ _('Field names') }</div>
+                    <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaFieldNames" placeholder="${ _('The list of fields to consume, e.g. orders,returns') }">
+                  </label>
+                  <label class="control-label"><div>${ _('Field types') }</div>
+                    <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.kafkaFieldTypes" placeholder="${ _('The list of field typs, e.g. string,int') }">
+                  </label>
+                </label>
+
+                <div class="control-group" data-bind="visible: createWizard.source.hasStreamSelected">
+                  <button class="btn" data-bind="click: createWizard.source.streamCheckConnection">
+                    ${_('Test')}
+                  </button>
+                </div>
+                </div>
               </div>
             <!-- /ko -->
 
@@ -366,21 +392,17 @@ ${ assist.assistPanel() }
                   <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamToken">
                 </label>
 
-                <label class="control-label"><div>${ _('Username') }</div>
+                <label class="control-label"><div>${ _('End point URL') }</div>
                   <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamEndpointUrl">
                 </label>
 
+                <br/>
+
                 <label class="control-label"><div>${ _('Object') }</div>
                   <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamObject">
                 </label>
               </div>
             <!-- /ko -->
-
-            <div class="control-group" data-bind="visible: createWizard.source.hasStreamSelected">
-              <button class="btn" data-bind="click: createWizard.source.streamCheckConnection">
-                ${_('Test')}
-              </button>
-            </div>
           <!-- /ko -->
 
           <div class="control-group" data-bind="visible: createWizard.source.inputFormat() == 'table'">
@@ -1377,6 +1399,7 @@ ${ assist.assistPanel() }
       self.isObjectStore.subscribe(function(newVal) {
         wizard.destination.useDefaultLocation(!newVal);
       });
+
       // Rdbms
       self.rdbmsMode = ko.observable('customRdbms');
       self.rdbmsMode.subscribe(function (val) {
@@ -1512,6 +1535,7 @@ ${ assist.assistPanel() }
           }
         });
       };
+
       // Table
       self.table = ko.observable('');
       self.tableName = ko.computed(function() {
@@ -1552,7 +1576,8 @@ ${ assist.assistPanel() }
           viewModel.createWizard.guessFieldTypes();
         }
       });
-      self.kafkaFieldType = ko.observable('delimited'); // delimited, bytearray
+      self.kafkaSchemaManual = ko.observable('manual');
+      self.kafkaFieldType = ko.observable('delimited');
       self.kafkaFieldDelimiter = ko.observable(',');
       self.kafkaFieldNames = ko.observable('');
       self.kafkaFieldTypes = ko.observable('');
@@ -1567,6 +1592,14 @@ ${ assist.assistPanel() }
         return (self.streamSelection() == 'kafka' && self.kafkaSelectedTopics()) ||
            (self.streamSelection() == 'sfdc')
       });
+      self.hasStreamSelected.subscribe(function(newValue) {
+        if (newValue) {
+          wizard.guessFormat();
+          if (newValue == 'kafka') {
+            wizard.destination.tableFormat('kudu');
+          }
+        }
+      });
       self.streamCheckConnection = function() {
         $.post("${ url('indexer:get_db_component') }", {
           "source": ko.mapping.toJSON(self)
@@ -1732,7 +1765,7 @@ ${ assist.assistPanel() }
           if (format.value == 'database' && wizard.source.inputFormat() != 'manual') {
             return false;
           }
-          if (format.value == 'file' && ['manual', 'rdbms', 'kafka'].indexOf(wizard.source.inputFormat()) == -1) {
+          if (format.value == 'file' && ['manual', 'rdbms', 'stream'].indexOf(wizard.source.inputFormat()) == -1) {
             return false;
           }
           else if (format.value == 'index' && ['file', 'query', 'table'].indexOf(wizard.source.inputFormat()) == -1) {

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