Browse Source

[flink] Adding Flink SQL examples

Romain Rigaux 5 years ago
parent
commit
33654c7a00

+ 1 - 1
apps/about/src/about/templates/admin_wizard.mako

@@ -109,7 +109,7 @@ ${ layout.menubar(section='quick_start') }
 
 
               % if has_connectors():
               % if has_connectors():
                 <!-- ko foreach: connectors -->
                 <!-- ko foreach: connectors -->
-                  <!-- ko if: ['hive', 'impala', 'mysql', 'postgresql', 'presto', 'phoenix'].indexOf(dialect) != -1 -->
+                  <!-- ko if: ['hive', 'impala', 'mysql', 'postgresql', 'presto', 'phoenix', 'flink'].indexOf(dialect) != -1 -->
                   <li>
                   <li>
                     <a href="javascript:void(0)" data-bind="click: $root.installConnectorDataExample">
                     <a href="javascript:void(0)" data-bind="click: $root.installConnectorDataExample">
                       <i class="fa fa-download"></i> <span data-bind="text: name"></span>
                       <i class="fa fa-download"></i> <span data-bind="text: name"></span>

+ 20 - 0
apps/beeswax/data/queries.json

@@ -241,5 +241,25 @@
       "file_resources": [],
       "file_resources": [],
       "settings": []
       "settings": []
     }
     }
+  },
+  {
+    "name": "Query and live display a live stream of data",
+    "desc": "Simple select of auto generated data or via the user table backed by a Kafka topic",
+    "dialects": ["flink"],
+    "type": "2",
+    "auto_load_only": false,
+    "data": {
+      "query": {
+        "query": "\nCREATE TABLE datagen (\n  f_sequence INT,\n  f_random INT,\n  f_random_str STRING,\n  ts AS localtimestamp,\n  WATERMARK FOR ts AS ts\n) WITH (\n  'connector' = 'datagen',\n  'rows-per-second'='5',\n  'fields.f_sequence.kind'='sequence',\n  'fields.f_sequence.start'='1',\n  'fields.f_sequence.end'='1000',\n  'fields.f_random.min'='1',\n  'fields.f_random.max'='1000',\n  'fields.f_random_str.length'='10'\n)\n;\n\nSELECT *\nFROM datagen\nLIMIT 50\n;\n\n\n\nCREATE TABLE user_behavior (\n  user_id BIGINT,\n  item_id BIGINT,\n  category_id BIGINT,\n  behavior STRING,\n  ts TIMESTAMP(3),\n  proctime AS PROCTIME(),   -- generates processing-time attribute using computed column\n  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND  -- defines watermark on ts column, marks ts as event-time attribute\n) WITH (\n  'connector' = 'kafka',  -- using kafka connector\n  'topic' = 'user_behavior',  -- kafka topic\n  'scan.startup.mode' = 'earliest-offset',  -- reading from the beginning\n  'properties.bootstrap.servers' = 'kafka:9094',  -- kafka broker address\n  'format' = 'json'  -- the data format is json\n)\n;\n\nSELECT * \nFROM user_behavior \nLIMIT 50\n;\n\n\nSELECT\n  HOUR(TUMBLE_START(ts, INTERVAL '1' HOUR)) as hour_of_day,\n  COUNT(*) as buy_cnt\nFROM\n  user_behavior\nWHERE\n  behavior = 'buy'\nGROUP BY\n  TUMBLE(ts, INTERVAL '1' HOUR)\n;\n  ",
+        "type": 0,
+        "email_notify": false,
+        "is_parameterized": false,
+        "database": ""
+      },
+      "functions": [],
+      "VERSION": "0.4.1",
+      "file_resources": [],
+      "settings": []
+    }
   }
   }
 ]
 ]

+ 9 - 0
apps/beeswax/data/tables.json

@@ -17,4 +17,13 @@
     "transactional": true,
     "transactional": true,
     "is_multi_inserts": true
     "is_multi_inserts": true
   }
   }
+  ,
+  {
+    "data_file": null,
+    "create_sql": "CREATE TABLE datagen ( f_sequence INT, f_random INT, f_random_str STRING, ts AS localtimestamp, WATERMARK FOR ts AS ts ) WITH ( 'connector' = 'datagen', 'rows-per-second'='5', 'fields.f_sequence.kind'='sequence', 'fields.f_sequence.start'='1', 'fields.f_sequence.end'='1000', 'fields.f_random.min'='1', 'fields.f_random.max'='1000', 'fields.f_random_str.length'='10' )\n",
+    "insert_sql": null,
+    "table_name": "datagen",
+    "dialects": ["flink"],
+    "transactional": true
+  }
 ]
 ]

+ 6 - 5
apps/beeswax/src/beeswax/management/commands/beeswax_install_examples.py

@@ -170,6 +170,7 @@ class SampleTable(object):
     else:
     else:
       self.partition_files = None
       self.partition_files = None
       self.filename = data_dict['data_file']
       self.filename = data_dict['data_file']
+    self._contents_file = None
     self.create_sql = data_dict['create_sql'].strip()
     self.create_sql = data_dict['create_sql'].strip()
     self.insert_sql = data_dict.get('insert_sql')
     self.insert_sql = data_dict.get('insert_sql')
     self.dialect = dialect
     self.dialect = dialect
@@ -187,7 +188,7 @@ class SampleTable(object):
         filepath = os.path.join(self._data_dir, filename)
         filepath = os.path.join(self._data_dir, filename)
         self.partition_files[partition_spec] = filepath
         self.partition_files[partition_spec] = filepath
         self._check_file_contents(filepath)
         self._check_file_contents(filepath)
-    else:
+    elif self.filename:
       self._contents_file = os.path.join(self._data_dir, self.filename)
       self._contents_file = os.path.join(self._data_dir, self.filename)
       self._check_file_contents(self._contents_file)
       self._check_file_contents(self._contents_file)
 
 
@@ -205,11 +206,12 @@ class SampleTable(object):
     if self.partition_files:
     if self.partition_files:
       for partition_spec, filepath in list(self.partition_files.items()):
       for partition_spec, filepath in list(self.partition_files.items()):
         self.load_partition(django_user, partition_spec, filepath, columns=self.columns)
         self.load_partition(django_user, partition_spec, filepath, columns=self.columns)
-    else:
+    elif self._contents_file:
       self.load(django_user)
       self.load(django_user)
 
 
     return True
     return True
 
 
+
   def create(self, django_user):
   def create(self, django_user):
     """
     """
     Create SQL sample table.
     Create SQL sample table.
@@ -413,8 +415,7 @@ class SampleQuery(object):
       query = SavedQuery.objects.get(owner=django_user, name=self.name, type=self.type)
       query = SavedQuery.objects.get(owner=django_user, name=self.name, type=self.type)
     except SavedQuery.DoesNotExist:
     except SavedQuery.DoesNotExist:
       query = SavedQuery(owner=django_user, name=self.name, type=self.type, desc=self.desc)
       query = SavedQuery(owner=django_user, name=self.name, type=self.type, desc=self.desc)
-      # The data field needs to be a string. The sample file writes it
-      # as json (without encoding into a string) for readability.
+      # The data field needs to be a string. The sample file writes it as json (without encoding into a string) for readability.
       query.data = json.dumps(self.data)
       query.data = json.dumps(self.data)
       query.save()
       query.save()
       LOG.info('Successfully installed sample design: %s' % (self.name,))
       LOG.info('Successfully installed sample design: %s' % (self.name,))
@@ -454,7 +455,7 @@ class SampleQuery(object):
 
 
       # Share with default group
       # Share with default group
       examples_dir.share(django_user, Document2Permission.READ_PERM, groups=[get_default_user_group()])
       examples_dir.share(django_user, Document2Permission.READ_PERM, groups=[get_default_user_group()])
-      LOG.info('Successfully installed sample query: %s' % (self.name,))
+      LOG.info('Successfully installed sample query: %s' % doc2)
 
 
 
 
   def _document_type(self, type, interpreter=None):
   def _document_type(self, type, interpreter=None):

+ 1 - 1
desktop/core/src/desktop/lib/connectors/models.py

@@ -175,7 +175,7 @@ def _augment_connector_properties(connector):
 
 
   for connector_type in get_connectors_types():
   for connector_type in get_connectors_types():
     if connector_type['dialect'] == connector['dialect'] and connector['interface'] and \
     if connector_type['dialect'] == connector['dialect'] and connector['interface'] and \
-          connector_type.get('interface') == connector.get('interface'):
+        connector_type.get('interface') == connector.get('interface'):
       connector_types.insert(0, connector_type)
       connector_types.insert(0, connector_type)
       break
       break
     if connector_type['dialect'] == connector['dialect']:
     if connector_type['dialect'] == connector['dialect']:

+ 3 - 0
desktop/core/src/desktop/models.py

@@ -1189,6 +1189,9 @@ class Document2(models.Model):
     res = '%s - %s - %s - %s' % (force_unicode(self.name), self.owner, self.type, self.uuid)
     res = '%s - %s - %s - %s' % (force_unicode(self.name), self.owner, self.type, self.uuid)
     return force_unicode(res)
     return force_unicode(res)
 
 
+  def __str__(self):
+    return self.__unicode__()
+
   @property
   @property
   def data_dict(self):
   def data_dict(self):
     if not self.data:
     if not self.data: