Browse Source

HUE-8208 [importer] Generate Envelope Solr output

Romain Rigaux 7 years ago
parent
commit
b4e36444b9

+ 1 - 0
desktop/libs/indexer/src/indexer/api3.py

@@ -519,6 +519,7 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
     elif destination['outputFormat'] == 'file':
       properties['path'] = file_format["path"]
       properties['format'] = file_format['tableFormat'] # or csv
+      
 
   properties["app_name"] = 'Data Ingest'
   properties["inputFormat"] = file_format['inputFormat']

+ 13 - 2
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -107,7 +107,7 @@ class EnvelopeIndexer(object):
                 auth-endpoint = "%(streamEndpointUrl)s"
               }
             }
-  """
+  """ % properties
       else:
         raise PopupException(_('Stream format of %(inputFormat)s not recognized: %(streamSelection)s') % properties)
     elif properties['inputFormat'] == 'file':
@@ -155,8 +155,19 @@ class EnvelopeIndexer(object):
               type = hive
               table.name = "%(output_table)s"
           }""" % properties
+    elif properties['ouputFormat'] == 'index':
+      output = """dependencies = [inputdata]
+        planner {
+            type = upstert
+        }
+        output {
+            type = solr
+            connection = "%(connection)s"
+            collection.name = "%(collectionName)s"
+        }
+        }""" % properties
     else:
-      raise PopupException(_('Input format not recognized: %(inputFormat)s') % properties)
+      raise PopupException(_('Output format not recognized: %(ouputFormat)s') % properties)
       
     return """
 application {

+ 62 - 6
desktop/libs/indexer/src/indexer/indexers/envelope_tests.py

@@ -103,13 +103,13 @@ def test_generate_from_stream_sfdc_to_hive_table():
         input {
             type = sfdc
             mode = fetch-all
-            sobject = %(streamObject)s
+            sobject = Opportunities
             sfdc: {
               partner: {
-                username = "%(streamUsername)s"
-                password = "%(streamPassword)s"
-                token = "%(streamToken)s"
-                auth-endpoint = "%(streamEndpointUrl)s"
+                username = "test"
+                password = "test"
+                token = "token"
+                auth-endpoint = "http://sfdc/api"
               }
             }
   
@@ -126,4 +126,60 @@ def test_generate_from_stream_sfdc_to_hive_table():
               table.name = "sfdc"
           }
     }
-}''' in  config, config)
+}''' in  config, config)
+  
+
+def test_generate_from_stream_kafka_to_solr_index():
+  properties = {
+    'app_name': 'Ingest',
+
+    'inputFormat': 'stream',
+    'streamSelection': 'kafka',
+    'brokers': 'broker:9092',
+    'topics': 'kafkaTopic',
+    'kafkaFieldType': 'delimited',
+    'kafkaFieldDelimiter': ',',
+    'kafkaFieldNames': 'id,name',
+    'kafkaFieldTypes': 'int,string',
+
+    'ouputFormat': 'index',
+    'connection': 'http://self-service-analytics.hue.com:8983/solr/',
+    'collectionName': 'traffic'
+  }
+
+  config = EnvelopeIndexer(username='test').generate_config(properties)
+
+  assert_true('''steps {
+    inputdata {
+        input {
+            type = kafka
+                brokers = "broker:9092"
+                topics = kafkaTopic
+                encoding = string
+                translator {
+                    type = delimited
+                    delimiter = ","
+                    field.names = [id,name]
+                    field.types = [int,string]
+                }
+                window {
+                    enabled = true
+                    milliseconds = 60000
+                }
+        
+        }
+    }
+
+    outputdata {
+        dependencies = [inputdata]
+        planner {
+            type = upstert
+        }
+        output {
+            type = solr
+            connection = "http://self-service-analytics.hue.com:8983/solr/"
+            collection.name = "traffic"
+        }
+        }
+    }
+}''' in  config, config)

+ 2 - 2
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -331,7 +331,7 @@ ${ assist.assistPanel() }
             <!-- ko if: createWizard.source.streamSelection() == 'kafka' -->
               <div class="control-group">
                 <label class="control-label"><div>${ _('Topics') }</div>
-                  <select data-bind="options: createWizard.source.kafkaTopics,
+                  <select class="input-xxlarge" data-bind="options: createWizard.source.kafkaTopics,
                          value: createWizard.source.kafkaSelectedTopics,
                          optionsCaption: 'Choose...'"
                          placeholder="${ _('The list of topics to consume, e.g. orders,returns') }"></select>
@@ -1793,7 +1793,7 @@ ${ assist.assistPanel() }
           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) {
+          else if (format.value == 'index' && ['file', 'query', 'table', 'stream'].indexOf(wizard.source.inputFormat()) == -1) {
             return false;
           }
           if (format.value == 'hbase' && wizard.source.inputFormat() != 'rdbms') {