Browse Source

HUE-8298 [editor] Basic Kafka SQL connector via Spark Streaming

Romain Rigaux 7 years ago
parent
commit
3fd99205b3

+ 7 - 3
desktop/libs/metadata/src/metadata/kafka_api.py

@@ -94,9 +94,13 @@ def get_topics():
   if has_kafka_api():
     return KafkaApi().topics()
   else:
-    manager = ManagerApi()
-    broker_host = manager.get_kafka_brokers().split(',')[0].split(':')[0]
-    return [name for name in manager.get_kafka_topics(broker_host).keys() if not name.startswith('__')]
+    try:
+      manager = ManagerApi()
+      broker_host = manager.get_kafka_brokers().split(',')[0].split(':')[0]
+      return [name for name in manager.get_kafka_topics(broker_host).keys() if not name.startswith('__')]
+    except Exception, e:
+      print e
+      return ["traffic", "hueAccessLogs"]
 
 
 def get_topic(name):

+ 1 - 2
desktop/libs/metadata/src/metadata/kafka_client.py

@@ -61,5 +61,4 @@ class KafkaApi(object):
       response = self._root.get('topics')
       return json.loads(response)
     except RestException, e:
-      return ['traffic', 'jiras']
-#       raise KafkaApiException(e)
+      raise KafkaApiException(e)

+ 49 - 1
desktop/libs/notebook/src/notebook/connectors/kafka.py

@@ -18,6 +18,7 @@
 from __future__ import absolute_import
 
 import logging
+import json
 
 from django.core.urlresolvers import reverse
 from django.utils.translation import ugettext as _
@@ -48,7 +49,54 @@ class KafkaApi(Api):
     response = {}
 
     try:
-      response['databases'] = get_topics()
+      if database is None:
+        response['databases'] = ['default']
+      elif table is None:
+        response['tables_meta'] = get_topics()
+      else:
+        response = {
+          u'status': 0,
+          u'comment': u'test test test 22',
+          u'hdfs_link': u'/filebrowser/view=/user/hive/warehouse/web_logs',
+          u'extended_columns': [
+            {u'comment': u'', u'type': u'bigint', u'name': u'_version_'},
+            {u'comment': u'The app', u'type': u'string', u'name': u'app'},
+            {u'comment': u'test test   test 22', u'type': u'smallint', u'name': u'bytes'},
+            {u'comment': u'The citi', u'type': u'string', u'name': u'city'},
+            {u'comment': u'', u'type': u'string', u'name': u'client_ip'},
+            {u'comment': u'', u'type': u'tinyint', u'name': u'code'},
+            {u'comment': u'', u'type': u'string', u'name': u'country_code'},
+            {u'comment': u'', u'type': u'string', u'name': u'country_code3'},
+            {u'comment': u'', u'type': u'string', u'name': u'country_name'},
+            {u'comment': u'', u'type': u'string', u'name': u'device_family'},
+            {u'comment': u'', u'type': u'string', u'name': u'extension'},
+            {u'comment': u'', u'type': u'float', u'name': u'latitude'},
+            {u'comment': u'', u'type': u'float', u'name': u'longitude'},
+            {u'comment': u'', u'type': u'string', u'name': u'method'},
+            {u'comment': u'', u'type': u'string', u'name': u'os_family'},
+            {u'comment': u'', u'type': u'string', u'name': u'os_major'},
+            {u'comment': u'', u'type': u'string', u'name': u'protocol'},
+            {u'comment': u'', u'type': u'string', u'name': u'record'},
+            {u'comment': u'', u'type': u'string', u'name': u'referer'},
+            {u'comment': u'', u'type': u'bigint', u'name': u'region_code'},
+            {u'comment': u'', u'type': u'string', u'name': u'request'},
+            {u'comment': u'', u'type': u'string', u'name': u'subapp'},
+            {u'comment': u'', u'type': u'string', u'name': u'time'},
+            {u'comment': u'', u'type': u'string', u'name': u'url'},
+            {u'comment': u'', u'type': u'string', u'name': u'user_agent'},
+            {u'comment': u'', u'type': u'string', u'name': u'user_agent_family'},
+            {u'comment': u'', u'type': u'string', u'name': u'user_agent_major'},
+            {u'comment': u'', u'type': u'string', u'name': u'id'},
+            {u'comment': u'', u'type': u'string', u'name': u'date'}
+          ],
+          u'support_updates': False,
+          u'partition_keys': [
+            {u'type': u'string', u'name': u'date'}
+          ],
+          u'columns': [u'_version_', u'app', u'bytes', u'city', u'client_ip', u'code', u'country_code', u'country_code3', u'country_name', u'device_family', u'extension', u'latitude', u'longitude', u'method', u'os_family', u'os_major', u'protocol', u'record', u'referer', u'region_code', u'request', u'subapp', u'time', u'url', u'user_agent', u'user_agent_family', u'user_agent_major', u'id', u'date'],
+          u'is_view': False
+        }
+
     except Exception, e:
       LOG.warn('Autocomplete data fetching error: %s' % e)
       response['code'] = 500

+ 7 - 0
desktop/libs/notebook/src/notebook/templates/editor_components.mako

@@ -187,6 +187,7 @@ ${ sqlSyntaxDropdown.sqlSyntaxDropdown() }
                 <!-- ko case: 'jar' -->Spark Submit Jar<!-- /ko -->
                 <!-- ko case: 'py' -->Spark Submit Python<!-- /ko -->
                 <!-- ko case: 'solr' -->Solr SQL<!-- /ko -->
+                <!-- ko case: 'kafkasql' -->Kafka SQL<!-- /ko -->
                 <!-- ko case: 'markdown' -->Markdown<!-- /ko -->
                 <!-- ko case: 'text' -->Text<!-- /ko -->
                 <!-- ko case: $default -->SQL<!-- /ko -->
@@ -2855,6 +2856,12 @@ function togglePresentation(value) {};
           snippetIcon: 'fa-database',
           sqlDialect: true
         },
+        kafkasql: {
+          placeHolder: '${ _("Example: SELECT fieldA, FieldB FROM collectionname, or press CTRL + space") }',
+          aceMode: 'ace/mode/mysql',
+          snippetIcon: 'fa-database',
+          sqlDialect: true
+        },
         java : {
           snippetIcon: 'fa-file-code-o'
         },