Browse Source

HUE-8208 [importer] Generate proper import of SFDC to a Table

Romain Rigaux 7 years ago
parent
commit
5609584f99

+ 6 - 2
desktop/libs/indexer/src/indexer/api3.py

@@ -510,8 +510,12 @@ 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
-      properties["output_table"] = "impala::%s" % collection_name
-      properties["kudu_master"] = manager.get_kudu_master()
+      if file_format['inputFormat'] == 'stream':
+        properties["output_table"] = "impala::%s" % collection_name
+        properties["kudu_master"] = manager.get_kudu_master()
+      else:
+        properties['output_table'] = collection_name
+
     elif destination['outputFormat'] == 'file':
       properties['path'] = file_format["path"]
       properties['format'] = file_format['tableFormat'] # or csv

+ 42 - 32
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -97,16 +97,16 @@ class EnvelopeIndexer(object):
         """ % properties
       elif properties['streamSelection'] == 'sfdc':
         input = """type = sfdc
-        mode = fetch-all
-        sobject = %(streamObject)s
-        sfdc: {
-          partner: {
-            username = "%(streamUsername)s"
-            password = "%(streamPassword)s"
-            token = "%(streamToken)s"
-            auth-endpoint = "%(streamEndpointUrl)s"
-          }
-        }
+            mode = fetch-all
+            sobject = %(streamObject)s
+            sfdc: {
+              partner: {
+                username = "%(streamUsername)s"
+                password = "%(streamPassword)s"
+                token = "%(streamToken)s"
+                auth-endpoint = "%(streamEndpointUrl)s"
+              }
+            }
   """
       else:
         raise PopupException(_('Stream format of %(inputFormat)s not recognized: %(streamSelection)s') % properties)
@@ -121,30 +121,40 @@ class EnvelopeIndexer(object):
 
     if properties['ouputFormat'] == 'file':
       output = """dependencies = [inputdata]
-    planner = {
-      type = overwrite
-    }
-    output = {
-      type = filesystem
-      path = %(path)s
-      format = %(format)s
-      header = true
-    }""" % properties
-    elif properties['ouputFormat'] == 'table':
-      output = """dependencies = [inputdata]
-        deriver {
-            type = sql
-            query.literal = \"""
-                SELECT measurement_time, number_of_vehicles FROM inputdata\"""
+        planner = {
+          type = overwrite
         }
-        planner {
-            type = upsert
-        }
-        output {
-            type = kudu
-            connection = "%(kudu_master)s"
-            table.name = "%(output_table)s"
+        output = {
+          type = filesystem
+          path = %(path)s
+          format = %(format)s
+          header = true
         }""" % properties
+    elif properties['ouputFormat'] == 'table':
+      if properties['inputFormat'] == 'stream' and properties['streamSelection'] == 'kafka':
+        output = """dependencies = [inputdata]
+          deriver {
+              type = sql
+              query.literal = \"""
+                  SELECT measurement_time, number_of_vehicles FROM inputdata\"""
+          }
+          planner {
+              type = upsert
+          }
+          output {
+              type = kudu
+              connection = "%(kudu_master)s"
+              table.name = "%(output_table)s"
+          }""" % properties
+      else:
+        output = """dependencies = [inputdata]
+          planner {
+              type = append
+          }
+          output {
+              type = hive
+              table.name = "%(output_table)s"
+          }""" % properties
     else:
       raise PopupException(_('Input format not recognized: %(inputFormat)s') % properties)
       

+ 81 - 27
desktop/libs/indexer/src/indexer/indexers/envelope_tests.py

@@ -26,9 +26,14 @@ def test_generate_from_kafka_to_file_csv():
   properties = {
     'app_name': 'Ingest',
 
-    'inputFormat': 'kafka',
+    'inputFormat': 'stream',
+    'streamSelection': 'kafka',
     'brokers': 'broker:9092',
     'topics': 'kafkaTopic',
+    'kafkaFieldType': 'delimited',
+    'kafkaFieldDelimiter': ',',
+    'kafkaFieldNames': 'id,name',
+    'kafkaFieldTypes': 'int,string',
 
     'ouputFormat': 'file',
     'path': '/tmp/output',
@@ -37,39 +42,88 @@ def test_generate_from_kafka_to_file_csv():
 
   config = EnvelopeIndexer(username='test').generate_config(properties)
 
-  assert_true('''application {
-    name = Ingest
-    batch.milliseconds = 5000
-    executors = 1
-    executor.cores = 1
-    executor.memory = 1G
+  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 = overwrite
+        }
+        output = {
+          type = filesystem
+          path = /tmp/output
+          format = csv
+          header = true
+        }
+    }
 }
+''' in  config, config)
 
-steps {
+
+def test_generate_from_stream_sfdc_to_hive_table():
+  properties = {
+    'app_name': 'Ingest',
+
+    'inputFormat': 'stream',    
+    'streamSelection': 'sfdc',
+    'streamUsername': 'test',
+    'streamPassword': 'test',
+    'streamToken': 'token',
+    'streamEndpointUrl': 'http://sfdc/api',
+    'streamObject': 'Opportunities',
+
+    'ouputFormat': 'table',
+    'output_table': 'sfdc',
+    'format': 'text'
+  }
+
+  config = EnvelopeIndexer(username='test').generate_config(properties)
+
+  assert_true('''steps {
     inputdata {
         input {
-              type = kafka
-              brokers = "broker:9092"
-              topics = kafkaTopic
-              encoding = string
-              window {
-                  enabled = true
-                  milliseconds = 30000
+            type = sfdc
+            mode = fetch-all
+            sobject = %(streamObject)s
+            sfdc: {
+              partner: {
+                username = "%(streamUsername)s"
+                password = "%(streamPassword)s"
+                token = "%(streamToken)s"
+                auth-endpoint = "%(streamEndpointUrl)s"
               }
-
+            }
+  
         }
     }
 
     outputdata {
-    dependencies = [inputdata]
-    planner = {
-      type = overwrite
-    }
-    output = {
-      type = filesystem
-      path = %(path)s
-      format = %(format)s
-      header = true
-    }
+        dependencies = [inputdata]
+          planner {
+              type = append
+          }
+          output {
+              type = hive
+              table.name = "sfdc"
+          }
     }
-}''' in  config, config)
+}''' in  config, config)

+ 7 - 3
desktop/libs/indexer/src/indexer/templates/importer.mako

@@ -1364,9 +1364,13 @@ ${ assist.assistPanel() }
         self.path('');
         resizeElements();
         self.rdbmsMode('customRdbms');
-        if (val == 'stream' && self.streamSelection() == 'kafka') {
-          wizard.guessFormat();
-          wizard.destination.tableFormat('kudu');
+        if (val == 'stream') {
+          if (self.streamSelection() == 'kafka') {
+            wizard.guessFormat();
+            wizard.destination.tableFormat('kudu');
+          } else {
+            wizard.destination.tableFormat('text');
+          }
         }
       });
       self.inputFormatsAll = ko.observableArray([