Browse Source

HUE-8208 [importer] Skeleton to generate a preview of data from Kafka

Romain Rigaux 7 years ago
parent
commit
5d138a2b8f

+ 1 - 4
desktop/core/src/desktop/templates/assist.mako

@@ -2237,10 +2237,7 @@ from desktop.views import _ko
             <!-- ko if: isSolr -->
             ${ _('Indexes') }
             <!-- /ko -->
-            <!-- ko if: isKafka -->
-            ${ _('Streams') }
-            <!-- /ko -->
-            <!-- ko ifnot: isSolr || isKafka  -->
+            <!-- ko ifnot: isSolr -->
             ${ _('Tables') }
             <!-- ko if: statementCount() > 1 -->
             <div class="statement-count">${ _('Statement') } <span data-bind="text: activeStatementIndex() + '/' + statementCount()"></span></div>

+ 26 - 18
desktop/libs/indexer/src/indexer/api3.py

@@ -420,6 +420,7 @@ def _envelope_job(request, file_format, collection_name, start_time=None, lib_pa
   indexer = EnvelopeIndexer(request.user, request.fs)
 
   lib_path = '/tmp/envelope-0.5.0.jar'
+  input_path = None
 
   if file_format['inputFormat'] == 'table':
     db = dbms.get(request.user)
@@ -427,30 +428,37 @@ def _envelope_job(request, file_format, collection_name, start_time=None, lib_pa
     input_path = table_metadata.path_location
   elif file_format['inputFormat'] == 'file':
     input_path = '${nameNode}%s' % file_format["path"]
-  elif file_format['inputFormat'] == 'streams':
     properties = {
-      'streamSelection': file_format['streamSelection'],
-      'streamUsername': file_format['streamUsername'],
-      'streamPassword': file_format['streamPassword'],
-      'streamToken': file_format['streamToken'],
-      'streamEndpointUrl': file_format['streamEndpointUrl'],
-      'streamObject': file_format['streamObject'],
+      'format': 'json'
     }
-    input_path = None
-  else:
-    input_path = None
-
-    manager = ManagerApi()
+  elif file_format['inputFormat'] == 'stream':
+    if file_format['streamSelection'] == 'sfdc':
+      properties = {
+        'streamSelection': file_format['streamSelection'],
+        'streamUsername': file_format['streamUsername'],
+        'streamPassword': file_format['streamPassword'],
+        'streamToken': file_format['streamToken'],
+        'streamEndpointUrl': file_format['streamEndpointUrl'],
+        'streamObject': file_format['streamObject'],
+      }
+    elif file_format['streamSelection'] == 'kafka':
+      manager = ManagerApi()
+      properties = {
+        "brokers": manager.get_kafka_brokers(),
+        "output_table": "impala::%s" % collection_name,
+        "topics": file_format['kafkaSelectedTopics']
+      }
 
-    properties = {
-      "brokers": manager.get_kafka_brokers(),
-      "kudu_master": manager.get_kudu_master(),
-      "output_table": "impala::%s" % collection_name,
-      "topics": file_format['kafkaSelectedTopics']
-    }
+    if file_format['outputFormat'] == 'table':
+      properties["output_table"] = "impala::%s" % collection_name
+      properties["kudu_master"] = manager.get_kudu_master()
+    elif file_format['inputFormat'] == 'file':
+      properties['path'] = file_format["path"]
+      properties['format'] = file_format['tableFormat'] # or csv
 
   properties["inputFormat"] = file_format['inputFormat']
   properties["app_name"] = 'Data Ingest'
+
   morphline = indexer.generate_config(properties)
 
   return indexer.run(request, collection_name, morphline, input_path, start_time=start_time, lib_path=lib_path)

+ 27 - 27
desktop/libs/indexer/src/indexer/indexers/envelope.py

@@ -22,6 +22,7 @@ from django.core.urlresolvers import reverse
 from django.utils.translation import ugettext as _
 
 from notebook.models import make_notebook
+from desktop.lib.exceptions_renderable import PopupException
 
 
 LOG = logging.getLogger(__name__)
@@ -78,24 +79,24 @@ class EnvelopeIndexer(object):
 
   def generate_config(self, properties):
     if properties['inputFormat'] == 'kafka':
-      input = """            type = 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
-              translator {
-                  type = delimited
-                  delimiter = ","
-                  field.names = [measurement_time,number_of_vehicles]
-                  field.types = [long,int]
-              }
               window {
                   enabled = true
                   milliseconds = 60000
               }
       """ % properties
-    elif properties['inputFormat'] == 'streams':
-      if properties['streamSelection'] == 'SFDC':
-        input = """      type = sfdc
+    elif properties['inputFormat'] == 'stream':
+      if properties['streamSelection'] == 'sfdc':
+        input = """type = sfdc
         mode = fetch-all
         sobject = %(streamObject)s
         sfdc: {
@@ -107,30 +108,27 @@ class EnvelopeIndexer(object):
           }
         }
   """
-    else: # File
-      input = """      type = filesystem
-      path = example-input.json
-      format = json
+    elif properties['inputFormat'] == 'file':
+      input = """type = filesystem
+      path = %(path)s
+      format = %(format)s
       """ % properties
-      # header = false
-      
+    else:
+      raise PopupException(_('Input format not recognized: %(inputFormat)s') % properties)
+
     if properties['ouputFormat'] == 'file':
-      # parquet, 
-      output = """    dependencies = [inputdata]
-    deriver {
-      type = sql
-      query.literal = "SELECT * FROM inputdata"
-    }
+      output = """dependencies = [inputdata]
     planner = {
       type = overwrite
     }
     output = {
       type = filesystem
-      path = example-output
-      format = csv
-    }"""
-    else: # Table
-      output = """        dependencies = [inputdata]
+      path = %(path)s
+      format = %(format)s
+      header = true
+    }""" % properties
+    elif properties['ouputFormat'] == 'table':
+      output = """dependencies = [inputdata]
         deriver {
             type = sql
             query.literal = \"""
@@ -144,6 +142,8 @@ class EnvelopeIndexer(object):
             connection = "%(kudu_master)s"
             table.name = "%(output_table)s"
         }""" % properties
+    else:
+      raise PopupException(_('Input format not recognized: %(inputFormat)s') % properties)
       
     return """
 application {

+ 75 - 0
desktop/libs/indexer/src/indexer/indexers/envelope_tests.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# 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.
+
+from django.contrib.auth.models import User
+from nose.tools import assert_equal, assert_true
+
+from indexer.indexers.envelope import EnvelopeIndexer
+
+
+def test_generate_from_kafka_to_file_csv():
+  properties = {
+    'app_name': 'Ingest',
+
+    'inputFormat': 'kafka',
+    'brokers': 'broker:9092',
+    'topics': 'kafkaTopic',
+
+    'ouputFormat': 'file',
+    'path': '/tmp/output',
+    'format': 'csv'
+  }
+
+  config = EnvelopeIndexer(username='test').generate_config(properties)
+
+  assert_true('''application {
+    name = Ingest
+    batch.milliseconds = 5000
+    executors = 1
+    executor.cores = 1
+    executor.memory = 1G
+}
+
+steps {
+    inputdata {
+        input {
+              type = kafka
+              brokers = "broker:9092"
+              topics = kafkaTopic
+              encoding = string
+              window {
+                  enabled = true
+                  milliseconds = 30000
+              }
+
+        }
+    }
+
+    outputdata {
+    dependencies = [inputdata]
+    planner = {
+      type = overwrite
+    }
+    output = {
+      type = filesystem
+      path = %(path)s
+      format = %(format)s
+      header = true
+    }
+    }
+}''' in  config, config)