فهرست منبع

HUE-9084 [notebook] Adding a skeleton of Flink SQL interpreter

There is no execution yet, this is mostly plumbing.

e.g. To add in the hue.ini like:

  [notebook]
  [[interpreters]]
  [[[flink]]]
  name=Flink SQL
  interface=flink
  options='{"api_url": "http://flink:10000"}'
Romain 6 سال پیش
والد
کامیت
bc032d9737
2فایلهای تغییر یافته به همراه137 افزوده شده و 0 حذف شده
  1. 3 0
      desktop/libs/notebook/src/notebook/connectors/base.py
  2. 134 0
      desktop/libs/notebook/src/notebook/connectors/flink.py

+ 3 - 0
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -399,6 +399,9 @@ def get_api(request, snippet):
   elif interface == 'ksql':
     from notebook.connectors.ksql import KSqlApi
     return KSqlApi(request.user)
+  elif interface == 'flink':
+    from notebook.connectors.flink import FlinkSqlApi
+    return FlinkSqlApi(request.user)
   elif interface == 'kafka':
     from notebook.connectors.kafka import KafkaApi
     return KafkaApi(request.user)

+ 134 - 0
desktop/libs/notebook/src/notebook/connectors/flink.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python
+# 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 __future__ import absolute_import
+
+import logging
+
+from django.core.urlresolvers import reverse
+from django.utils.translation import ugettext as _
+
+from desktop.lib.i18n import force_unicode
+
+from notebook.connectors.base import Api, QueryError
+
+
+LOG = logging.getLogger(__name__)
+
+
+def query_error_handler(func):
+  def decorator(*args, **kwargs):
+    try:
+      return func(*args, **kwargs)
+    except Exception as e:
+      message = force_unicode(str(e))
+      raise QueryError(message)
+  return decorator
+
+
+class FlinkSqlApi(Api):
+
+  def __init__(self, user, interpreter=None):
+    Api.__init__(self, user, interpreter=interpreter)
+
+    self.options = interpreter['options']
+    self.db = FlinkSqlClient(user=user, api_url=self.options['api_url'])
+
+
+  @query_error_handler
+  def execute(self, notebook, snippet):
+
+    data, description = self.db.query(snippet['statement'])
+    has_result_set = data is not None
+
+    return {
+      'sync': True,
+      'has_result_set': has_result_set,
+      'result': {
+        'has_more': False,
+        'data': data if has_result_set else [],
+        'meta': [{
+          'name': col[0],
+          'type': col[1],
+          'comment': ''
+        } for col in description] if has_result_set else [],
+        'type': 'table'
+      }
+    }
+
+
+  @query_error_handler
+  def check_status(self, notebook, snippet):
+    return {'status': 'available'}
+
+
+  @query_error_handler
+  def autocomplete(self, snippet, database=None, table=None, column=None, nested=None):
+    response = {}
+
+    try:
+      if database is None:
+        response['databases'] = ['tables', 'topics', 'streams']
+      elif table is None:
+        if database == 'tables':
+          response['tables_meta'] = self.db.show_tables()
+        elif database == 'topics':
+          response['tables_meta'] = self.db.show_topics()
+        elif database == 'streams':
+          response['tables_meta'] = [
+            {'name': t['name'], 'type': t['type'], 'comment': 'Topic: %(topic)s Format: %(format)s' % t}
+            for t in self.db.show_streams()
+          ]
+      elif column is None:
+        columns = self.db.get_columns(table)
+        response['columns'] = [col['name'] for col in columns]
+        response['extended_columns'] = [{
+            'comment': col.get('comment'),
+            'name': col.get('name'),
+            'type': str(col['schema'].get('type'))
+          } for col in columns
+        ]
+      else:
+        response = {}
+
+    except Exception as e:
+      LOG.warn('Autocomplete data fetching error: %s' % e)
+      response['code'] = 500
+      response['error'] = e.message
+
+    return response
+
+
+class FlinkSqlClient():
+  def __init__(self, user, api_url):
+    pass
+
+  def query(self, statement):
+    LOG.info('Executing query: %s' % statement)
+    return [[1, 2], [3, 4]], [['col1', 'INT_TYPE'], ['col2', 'INT_TYPE']]  # Data rows, column rows
+
+  def show_tables(self):
+    return []
+
+  def show_topics(self):
+    return []
+
+  def show_streams(self):
+    return []
+
+  def get_columns(self, table):
+    return []