瀏覽代碼

HUE-9077 [ksql] Basic stream query support

Romain 6 年之前
父節點
當前提交
9edc526f72
共有 1 個文件被更改,包括 12 次插入5 次删除
  1. 12 5
      desktop/libs/kafka/src/kafka/ksql_client.py

+ 12 - 5
desktop/libs/kafka/src/kafka/ksql_client.py

@@ -80,10 +80,15 @@ class KSqlApi(object):
     data = []
     metadata = []
 
-    if statement.strip().lower().startswith('select'):
-      for line in self.client.query(statement):
-        data_line = json.loads(line)
-        data.append(data_line['row']['columns']) # TODO: streams, dataline['errorMessage']
+    if statement.strip().lower().startswith('select') or statement.strip().lower().startswith('print'):
+      # STREAMS requires a LIMIT currently or https://github.com/bryanyang0528/ksql-python/pull/60
+      result = self.client.query(statement)
+      for line in ''.join(list(result)).split('\n'):
+        # Until https://github.com/bryanyang0528/ksql-python/issues/57
+        # columns = line.keys()
+        # data.append([line[col] for col in columns])
+        data.append([line])
+      metadata = [['Row', 'STRING']]
     else:
       data, metadata = self._decode_result(
         self.ksql(statement)
@@ -96,7 +101,9 @@ class KSqlApi(object):
     columns = []
     data = []
 
-    if result['@type'] == 'kafka_topics':
+    if result['@type'] == 'statement_error':
+      raise KSqlApiException(result['message'])
+    elif result['@type'] == 'kafka_topics':
       columns = result['topics'][0].keys()
       for line in result['topics']:
         row = []