Jelajahi Sumber

[importer] Use connector when submiting creation task

Romain 5 tahun lalu
induk
melakukan
af23eeb8b1

+ 4 - 2
desktop/libs/indexer/src/indexer/api3.py

@@ -442,13 +442,15 @@ def importer_submit(request):
       request.fs
     )
 
-    job_handle = api.create_table_from_kafka(**args)
+    job_nb = api.create_table_from_kafka(**args)
 
     if request.POST.get('show_command'):
       job_handle = {
         'status': 0,
-        'commands': job_handle
+        'commands': job_nb
       }
+    else:
+      job_handle = job_nb.execute(request, batch=False)
   elif source['inputFormat'] == 'altus':
     # BDR copy or DistCP + DDL + Sentry DDL copy
     pass

+ 3 - 4
desktop/libs/indexer/src/indexer/indexers/flink_sql.py

@@ -52,7 +52,7 @@ class FlinkIndexer():
     final_table_name = table_name
 
     source_type = source['sourceType']
-    editor_type = '51'  # destination['sourceType']
+    editor_type = '55'  # destination['sourceType']
 
     sql = '''CREATE TABLE %(table_name)s (
     user_id BIGINT,
@@ -76,9 +76,8 @@ class FlinkIndexer():
     if dry_run:
       return sql
     else:
-      on_success_url = reverse(
-          'metastore:describe_table', kwargs={'database': database, 'table': final_table_name}
-      ) + '?source_type=' + source_type
+      on_success_url = reverse('metastore:describe_table', kwargs={'database': database, 'table': final_table_name}) + \
+          '?source_type=' + source_type
 
       return make_notebook(
           name=_('Creating table %(database)s.%(table)s') % {'database': database, 'table': final_table_name},

+ 4 - 0
desktop/libs/notebook/src/notebook/models.py

@@ -167,6 +167,10 @@ def make_notebook(
     ] if not is_notebook else []
   }
 
+  if has_connectors():  # To improve
+    data['dialect'] = interpreter['dialect']
+    data['type'] = 'flink-' + editor_connector
+
   if snippet_properties:
     data['snippets'][0]['properties'].update(snippet_properties)
   if result_properties: