Procházet zdrojové kódy

HUE-8574 [importer] Adding Flume flows

Romain Rigaux před 7 roky
rodič
revize
1f91325

+ 43 - 30
desktop/libs/indexer/src/indexer/api3.py

@@ -135,13 +135,15 @@ def guess_format(request):
   elif file_format['inputFormat'] == 'stream':
     if file_format['streamSelection'] == 'kafka':
       format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n", 'topics': get_topics()}
-    elif file_format['streamSelection'] == 'sfdc':
-      sf = Salesforce(
-          username=file_format['streamUsername'],
-          password=file_format['streamPassword'],
-          security_token=file_format['streamToken']
-      )
-      format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n", 'objects': [sobject['name'] for sobject in sf.restful('sobjects/')['sobjects'] if sobject['queryable']]}
+    elif file_format['streamSelection'] == 'flume':
+      format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n"}
+  elif file_format['inputFormat'] == 'sfdc':
+    sf = Salesforce(
+        username=file_format['streamUsername'],
+        password=file_format['streamPassword'],
+        security_token=file_format['streamToken']
+    )
+    format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n", 'objects': [sobject['name'] for sobject in sf.restful('sobjects/')['sobjects'] if sobject['queryable']]}    
 
   format_['status'] = 0
   return JsonResponse(format_)
@@ -284,7 +286,7 @@ def guess_field_types(request):
         "file": {
             "stream": stream,
             "name": file_format['path']
-          },
+        },
         "format": file_format['format']
       })
       type_mapping = dict(zip(kafkaFieldNames, kafkaFieldTypes))
@@ -292,26 +294,35 @@ def guess_field_types(request):
       for col in format_['columns']:
         col['keyType'] = type_mapping[col['name']]
         col['type'] = type_mapping[col['name']]
-    elif file_format['streamSelection'] == 'sfdc':
-      sf = Salesforce(
-          username=file_format['streamUsername'],
-          password=file_format['streamPassword'],
-          security_token=file_format['streamToken']
-      )
-      table_metadata = [{
-          'name': column['name'],
-          'type': column['type']
-        } for column in sf.restful('sobjects/%(streamObject)s/describe/' % file_format)['fields']
-      ]
-      query = 'SELECT %s FROM %s LIMIT 4' % (', '.join([col['name'] for col in table_metadata]), file_format['streamObject'])
-      print query
+    elif file_format['streamSelection'] == 'flume':
       format_ = {
-        "sample": [row.values()[1:] for row in sf.query_all(query)['records']],
-        "columns": [
-            Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
-            for col in table_metadata
-        ]
-       }
+          "sample": [['...']] * 4,
+          "columns": [
+              Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
+              for col in [{'name': 'message', 'type': 'string'}]
+          ]
+      }
+  elif file_format['streamSelection'] == 'sfdc':
+    sf = Salesforce(
+        username=file_format['streamUsername'],
+        password=file_format['streamPassword'],
+        security_token=file_format['streamToken']
+    )
+    table_metadata = [{
+        'name': column['name'],
+        'type': column['type']
+      } for column in sf.restful('sobjects/%(streamObject)s/describe/' % file_format)['fields']
+    ]
+    query = 'SELECT %s FROM %s LIMIT 4' % (', '.join([col['name'] for col in table_metadata]), file_format['streamObject'])
+    print query
+
+    format_ = {
+      "sample": [row.values()[1:] for row in sf.query_all(query)['records']],
+      "columns": [
+          Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
+          for col in table_metadata
+      ]
+    }
 
   return JsonResponse(format_)
 
@@ -351,7 +362,7 @@ def importer_submit(request):
     else:
       client = SolrClient(request.user)
       job_handle = _small_indexing(request.user, request.fs, client, source, destination, index_name)
-  elif source['inputFormat'] == 'stream' or destination['ouputFormat'] == 'stream':
+  elif source['inputFormat'] in ('stream', 'sfdc') or destination['ouputFormat'] == 'stream':
     job_handle = _envelope_job(request, source, destination, start_time=start_time, lib_path=destination['indexerJobLibPath'])
   elif destination['ouputFormat'] == 'database':
     job_handle = _create_database(request, source, destination, start_time)
@@ -523,8 +534,10 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
       'input_path': input_path,
       'format': 'csv'
     }
-  elif file_format['inputFormat'] == 'stream':
-    if file_format['streamSelection'] == 'sfdc':
+  elif file_format['inputFormat'] == 'stream' and file_format['streamSelection'] == 'flume':
+    pass
+  elif file_format['inputFormat'] in ('stream', 'sfdc'):
+    if file_format['inputFormat'] == 'sfdc':
       properties = {
         'streamSelection': file_format['streamSelection'],
         'streamUsername': file_format['streamUsername'],

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

@@ -117,6 +117,13 @@ ENABLE_KAFKA = Config(
   default=False
 )
 
+ENABLE_ENVELOPE = Config(
+  key="enable_envelope",
+  help=_t("Flag to turn on Envelope based jobs."),
+  type=bool,
+  default=False
+)
+
 ENABLE_ALTUS = Config(
   key="enable_altus",
   help=_t("Flag to turn on Altus imports."),

+ 113 - 34
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -22,7 +22,7 @@
   from filebrowser.conf import SHOW_UPLOAD_BUTTON
   from notebook.conf import ENABLE_SQL_INDEXER
 
-  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, ENABLE_KAFKA, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER, ENABLE_ALTUS
+  from indexer.conf import ENABLE_NEW_INDEXER, ENABLE_SQOOP, ENABLE_KAFKA, CONFIG_INDEXER_LIBS_PATH, ENABLE_SCALABLE_INDEXER, ENABLE_ALTUS, ENABLE_ENVELOPE
 %>
 
 <%namespace name="actionbar" file="actionbar.mako" />
@@ -348,37 +348,62 @@ ${ assist.assistPanel() }
               <div data-bind="template: { name: 'kafka-topic-template', data: $data }" class="margin-top-10 field inline-block"></div>
             <!-- /ko -->
 
-            <!-- ko if: createWizard.source.streamSelection() == 'sfdc' -->
+            <!-- ko if: createWizard.source.streamSelection() == 'flume' -->
               <div class="control-group">
-                <label class="control-label"><div>${ _('Username') }</div>
-                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamUsername" placeholder="user@company.com">
+ 
+                <label class="control-label"><div>${ _('Type') }</div>
+                  <select class="input-medium" data-bind="selectize: createWizard.source.channelSourceTypes, value: createWizard.source.channelSourceType, optionsText: 'name', optionsValue: 'value'"></select>
                 </label>
-
-                <label class="control-label"><div>${ _('Password') }</div>
-                  <input type="password" class="input-xxlarge" data-bind="value: createWizard.source.streamPassword">
-                </label>
-
-                <label class="control-label"><div>${ _('Token') }</div>
-                  <input type="password" class="input-xxlarge" data-bind="value: createWizard.source.streamToken">
+                <!-- ko if: ['directory', 'exec', 'syslogs'].indexOf(createWizard.source.channelSourceType()) != -1 -->
+                <label class="control-label"><div>${ _('Hosts') }</div>
+                  <select class="input-xxlarge" data-bind="selectize: createWizard.source.channelSourceHosts, selectedOptions: createWizard.source.channelSourceSelectedHosts" multiple="true"></select>
                 </label>
-
-                <label class="control-label"><div>${ _('End point URL') }</div>
-                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamEndpointUrl">
-                </label>
-
-                <br/>
-                <!-- ko if: createWizard.source.streamUsername() && createWizard.source.streamPassword() && createWizard.source.streamToken() -->
-                <label class="control-label"><div>${ _('Object') }</div>
-                  <select class="input-xxlarge" data-bind="options: createWizard.source.streamObjects,
-                        value: createWizard.source.streamObject,
-                        optionsCaption: '${ _("Choose...") }'"
-                        placeholder="${ _('The SFDC object to import, e.g. Account, Opportunity') }"></select>
+                <!-- /ko -->
+                <!-- ko if: createWizard.source.channelSourceType() == 'directory' -->
+                <label class="control-label"><div>${ _('Path') }</div>
+                  <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.channelSourcePath" placeholder="${ _('The path to watch and consume') }">
                 </label>
                 <!-- /ko -->
+      
+                <!-- ko if: createWizard.source.channelSourceType() -->
+                <input data-bind="click: function() { createWizard.source.channelSourceType(null); }" class="btn" value="${ _('Clear') }"/>
+                <!-- /ko -->
+
               </div>
             <!-- /ko -->
           <!-- /ko -->
 
+          <!-- ko if: createWizard.source.inputFormat() == 'sfdc' -->
+            <div class="control-group">
+              <label class="control-label"><div>${ _('Username') }</div>
+                <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamUsername" placeholder="user@company.com">
+              </label>
+
+              <label class="control-label"><div>${ _('Password') }</div>
+                <input type="password" class="input-xxlarge" data-bind="value: createWizard.source.streamPassword">
+              </label>
+
+              <label class="control-label"><div>${ _('Token') }</div>
+                <input type="password" class="input-xxlarge" data-bind="value: createWizard.source.streamToken">
+              </label>
+
+              <label class="control-label"><div>${ _('End point URL') }</div>
+                <input type="text" class="input-xxlarge" data-bind="value: createWizard.source.streamEndpointUrl">
+              </label>
+
+              <br/>
+              <!-- ko if: createWizard.source.streamUsername() && createWizard.source.streamPassword() && createWizard.source.streamToken() -->
+              <label class="control-label"><div>${ _('Object') }</div>
+                <select class="input-xxlarge" data-bind="options: createWizard.source.streamObjects,
+                      value: createWizard.source.streamObject,
+                      optionsCaption: '${ _("Choose...") }'"
+                      placeholder="${ _('The SFDC object to import, e.g. Account, Opportunity') }"></select>
+              </label>
+              <!-- /ko -->
+            </div>
+
+          <!-- /ko -->
+
           <div class="control-group" data-bind="visible: createWizard.source.inputFormat() == 'table'">
 
             <!-- ko foreach: $root.createWizard.source.tables -->
@@ -538,6 +563,27 @@ ${ assist.assistPanel() }
               <!-- /ko -->
             <!-- /ko -->
 
+            <!-- ko if: outputFormat() == 'flume' -->
+      
+            <h4>${ _('Sink') }</h4>
+            <div class="row-fluid">
+              <div>
+                <label class="control-label"><div>${ _('Type') }</div>
+                  <select class="input-medium" data-bind="selectize: channelSinkTypes, value: channelSinkType, optionsText: 'name', optionsValue: 'value'"></select>
+                </label>
+                <!-- ko if: channelSinkType() == 'solr' -->
+                <label class="control-label"><div>${ _('Collection') }</div>
+                  <select class="input-xxlarge" data-bind="selectize: ['logIndex', 'apacheLogs'], value: channelSinkPath"></select>
+                </label>
+                <!-- /ko -->
+      
+                <!-- ko if: channelSinkType() -->
+                <input data-bind="click: function() { channelSourceType(null); }" class="btn" value="${ _('Clear') }"/>
+                <!-- /ko -->
+              </div>
+            </div>
+            <!-- /ko -->
+
             <span class="help-inline muted" data-bind="visible: !isTargetExisting() && isTargetChecking()">
               <i class="fa fa-spinner fa-spin"></i>
             </span>
@@ -1420,7 +1466,6 @@ ${ assist.assistPanel() }
 
       var init = function () {
         self.type = ko.observable(typeName);
-
         var types = viewModel.createWizard.fileTypes;
 
         for (var i = 0; i < types.length; i++) {
@@ -1484,7 +1529,6 @@ ${ assist.assistPanel() }
         }
         self.path('');
         resizeElements();
-        self.rdbmsMode('customRdbms');
         if (val === 'stream') {
           if (self.streamSelection() === 'kafka') {
             wizard.guessFormat();
@@ -1494,22 +1538,28 @@ ${ assist.assistPanel() }
           }
         } else if (val === 'table') {
           wizard.destination.outputFormat('altus');
+        } else if (val == 'rdbms') {
+          self.rdbmsMode('customRdbms');
         }
       });
       self.inputFormatsAll = ko.observableArray([
           {'value': 'file', 'name': 'File'},
+          % if ENABLE_SQOOP.get():
+          {'value': 'rdbms', 'name': 'External Database'},
+          % endif
           % if ENABLE_KAFKA.get():
           {'value': 'stream', 'name': 'Stream'},
           % endif
           % if ENABLE_ALTUS.get():
           {'value': 'table', 'name': 'Table'},
           % endif
-          % if ENABLE_SQOOP.get():
-          {'value': 'rdbms', 'name': 'External Database'},
-          % endif
           % if ENABLE_SQL_INDEXER.get():
           {'value': 'query', 'name': 'SQL Query'},
           % endif
+          ## TODO: Rename to 'Connectors'
+          % if ENABLE_ENVELOPE.get():
+          {'value': 'sfdc', 'name': 'Connectors'},
+          % endif
           {'value': 'manual', 'name': 'Manually'}
           ##{'value': 'text', 'name': 'Paste Text'},
       ]);
@@ -1713,7 +1763,7 @@ ${ assist.assistPanel() }
       // Streams, Kafka
       self.publicStreams = ko.observable([
         {'value': 'kafka', 'name': 'Kafka Topics'},
-        {'value': 'sfdc', 'name': 'SFDC'}
+        {'value': 'flume', 'name': 'Flume Agent'}
       ]);
       self.streamSelection = ko.observable(self.publicStreams()[0]['value']);
 
@@ -1741,6 +1791,22 @@ ${ assist.assistPanel() }
       });
       self.kafkaFieldSchemaPath = ko.observable('');
 
+      self.channelSourceTypes = ko.observableArray([
+        {'name': '${ _("Directory or File") }', 'value': 'directory'},
+        {'name': '${ _("Program") }', 'value': 'exec'},
+        {'name': '${ _("Syslogs") }', 'value': 'syslogs'},
+        {'name': '${ _("HTTP") }', 'value': 'http'}
+      ]);
+      self.channelSourceType = ko.observable();
+      self.channelSourceHosts = ko.observableArray(['host1.com', 'host2.com', 'host3.com', 'host4.com']);
+      self.channelSourceSelectedHosts = ko.observableArray([]);
+      self.channelSourceSelectedHosts.subscribe(function(newVal) {
+        if (newVal) {
+          viewModel.createWizard.guessFieldTypes();
+        }
+      })
+      self.channelSourcePath = ko.observable('/var/log/hue/access.log');
+
       self.streamUsername = ko.observable('');
       self.streamPassword = ko.observable('');
       self.streamToken = ko.observable('');
@@ -1819,9 +1885,12 @@ ${ assist.assistPanel() }
         if (self.inputFormat() === 'stream') {
           if (self.streamSelection() === 'kafka') {
             return self.kafkaSelectedTopics() && self.kafkaSelectedTopics().length > 0;
+          } else if (self.streamSelection() === 'flume') {
+            return self.channelSourceSelectedHosts().length > 0;
           }
-          return self.streamSelection() === 'sfdc' &&
-              self.streamUsername().length > 0 &&
+        }
+        if (self.inputFormat() === 'sfdc') {
+          return self.streamUsername().length > 0 &&
               self.streamPassword().length > 0 &&
               self.streamToken().length > 0 &&
               self.streamEndpointUrl().length > 0 &&
@@ -2096,7 +2165,7 @@ ${ assist.assistPanel() }
       self.primaryKeys = ko.observableArray();
       self.primaryKeyObjects = ko.observableArray();
 
-      self.useFieldEditor = ko.observable(false);
+      self.useFieldEditor = ko.observable('${ ENABLE_KAFKA.get() }' == 'True');
       // TODO: Figure out the database to use for field editor autocomplete
       self.fieldEditorDatabase = ko.observable('default');
       // TODO: Do something with the editor value
@@ -2154,6 +2223,14 @@ ${ assist.assistPanel() }
       ]);
       self.rdbmsSplitByColumn = ko.observableArray();
 
+      // Flume
+      self.channelSinkTypes = ko.observableArray([
+        {'name': '${ _("This topic") }', 'value': 'kafka'},
+        {'name': '${ _("Solr") }', 'value': 'solr'},
+        {'name': '${ _("HDFS") }', 'value': 'hdfs'}
+      ]);
+      self.channelSinkType = ko.observable();
+      self.channelSinkPath = ko.observable();
     };
 
     var CreateWizard = function (vm) {
@@ -2319,12 +2396,14 @@ ${ assist.assistPanel() }
             if (self.source.inputFormat() === 'stream') {
               if (self.source.streamSelection() === 'kafka') {
                 self.source.kafkaTopics(resp['topics']);
-              } else if (self.source.streamSelection() === 'sfdc') {
+              } else if (self.source.streamSelection() === 'flume') {
                 self.source.streamObjects(resp['objects']);
               }
+            } else if (self.source.inputFormat() === 'sfdc') {
+              self.source.streamObjects(resp['objects']);
             }
 
-            if (self.source.inputFormat() !== 'stream' || self.source.streamSelection() !== 'sfdc') {
+            if (self.source.inputFormat() !== 'stream') {
               self.guessFieldTypes();
             }
           }

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

@@ -258,11 +258,6 @@ ${ assist.assistPanel() }
 
 
 <script type="text/html" id="list-index">
-  <!-- ko if: channelSourceType() -->
-    Used to populate <span data-bind="text: channelSinkType"></span> <span data-bind="text: channelSinkPath"></span>
-    from <span data-bind="text: channelSourceType"></span> <span data-bind="text: channelSourcePath"></span>
-    via <a href="" data-bind="click: function() { huePubSub.publish('toggle.jobs.panel'); }">jobs <i class="fa fa-external-link"></i></a>
-  <!-- /ko -->
   <br/>
   <br/>
 
@@ -270,7 +265,7 @@ ${ assist.assistPanel() }
     <li class="active"><a href="#index-overview" data-toggle="tab" data-bind="click: function(){ $root.tab('index-overview'); }">${_('Overview')}</a></li>
     <li><a href="#index-columns" data-toggle="tab" data-bind="click: function(){ $root.tab('index-columns'); }">${_('Partitions')} (<span data-bind="text: fields().length"></span>)</a></li>
     <li>
-      <a href="#index-channel" data-toggle="tab" data-bind="click: function(){ $root.tab('index-channel'); }">${_('Channel')} (<span data-bind="text: channelSourceType() ? 1 : 0"></span>)</a>
+      <a href="#index-consumer" data-toggle="tab" data-bind="click: function(){ $root.tab('index-consumer'); }">${_('Consumers')} (0)</a>
     </li>
     <li>
       <a href="#index-sample" data-toggle="tab" data-bind="click: function(){ $root.tab('index-sample'); }">${_('Permissions')} (2)</a>
@@ -320,47 +315,13 @@ ${ assist.assistPanel() }
       </div>
     </div>
 
-    <div class="tab-pane margin-top-10" id="index-channel">
+    <div class="tab-pane margin-top-10" id="index-consumer">
       <h4>${ _('Source') }</h4>
       <div class="row-fluid">
         <div>
           <label class="control-label"><div>${ _('Type') }</div>
-            <select class="input-medium" data-bind="selectize: channelSourceTypes, value: channelSourceType, optionsText: 'name', optionsValue: 'value'"></select>
-          </label>
-          <!-- ko if: ['directory', 'exec', 'syslogs'].indexOf(channelSourceType()) != -1 -->
-          <label class="control-label"><div>${ _('Hosts') }</div>
-            <select class="input-xxlarge" data-bind="selectize: channelSourceHosts, selectedOptions: channelSourceSelectedHosts" multiple="true"></select>
-          </label>
-          <!-- /ko -->
-          <!-- ko if: channelSourceType() == 'directory' -->
-          <label class="control-label"><div>${ _('Path') }</div>
-            <input type="text" class="input-xxlarge" data-bind="value: channelSourcePath" placeholder="${ _('The path to watch and consume') }">
-          </label>
-          <!-- /ko -->
-
-          <!-- ko if: channelSourceType() -->
-          <input data-bind="click: function() { channelSourceType(null); }" class="btn" value="${ _('Clear') }"/>
-          <!-- /ko -->
-        </div>
-      </div>
-
-      <br><br>
-
-      <h4>${ _('Sink') }</h4>
-      <div class="row-fluid">
-        <div>
-          <label class="control-label"><div>${ _('Type') }</div>
-            <select class="input-medium" data-bind="selectize: channelSinkTypes, value: channelSinkType, optionsText: 'name', optionsValue: 'value'"></select>
+            Type
           </label>
-          <!-- ko if: channelSinkType() == 'solr' -->
-          <label class="control-label"><div>${ _('Collection') }</div>
-            <select class="input-xxlarge" data-bind="selectize: ['logIndex', 'apacheLogs'], value: channelSinkPath"></select>
-          </label>
-          <!-- /ko -->
-
-          <!-- ko if: channelSinkType() -->
-          <input data-bind="click: function() { channelSourceType(null); }" class="btn" value="${ _('Clear') }"/>
-          <!-- /ko -->
         </div>
       </div>
     </div>
@@ -565,24 +526,6 @@ ${ assist.assistPanel() }
         $.totalStorage(userPrefix + '_kafka_topics_' + self.name() + '_kafkaFieldTypes', newValue)
       });
 
-      self.channelSourceTypes = ko.observableArray([
-        {'name': '${ _("Directory or File") }', 'value': 'directory'},
-        {'name': '${ _("Program") }', 'value': 'exec'},
-        {'name': '${ _("Syslogs") }', 'value': 'syslogs'},
-        {'name': '${ _("HTTP") }', 'value': 'http'}
-      ]);
-      self.channelSourceType = ko.observable();
-      self.channelSourceHosts = ko.observableArray(['host1.com', 'host2.com', 'host3.com', 'host4.com']);
-      self.channelSourceSelectedHosts = ko.observableArray([]);
-      self.channelSourcePath = ko.observable('/var/log/hue/access.log');
-
-      self.channelSinkTypes = ko.observableArray([
-        {'name': '${ _("This topic") }', 'value': 'kafka'},
-        {'name': '${ _("Solr") }', 'value': 'solr'}
-      ]);
-      self.channelSinkType = ko.observable();
-      self.channelSinkPath = ko.observable();
-
       self.sample = ko.observableArray();
       self.samplePreview = ko.pureComputed(function () {
         return self.sample().slice(0, 5)