Explorar o código

[importer] Generate Flink create table with proper columns and topic

Romain %!s(int64=5) %!d(string=hai) anos
pai
achega
26d8fe6b7a
Modificáronse 1 ficheiros con 11 adicións e 13 borrados
  1. 11 13
      desktop/libs/indexer/src/indexer/indexers/flink_sql.py

+ 11 - 13
desktop/libs/indexer/src/indexer/indexers/flink_sql.py

@@ -54,23 +54,21 @@ class FlinkIndexer():
     source_type = source['sourceType']
     editor_type = '55'  # destination['sourceType']
 
+    columns = destination['columns']
+
     sql = '''CREATE TABLE %(table_name)s (
-    user_id BIGINT,
-    item_id BIGINT,
-    category_id BIGINT,
-    behavior STRING,
-    ts TIMESTAMP(3),
-    proctime AS PROCTIME(),
-    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
+%(columns)s
 ) WITH (
-    'connector' = 'kafka',
-    'topic' = 'user_behavior',
-    'scan.startup.mode' = 'earliest-offset',
-    'properties.bootstrap.servers' = 'kafka:9094',
-    'format' = 'json'
+  'connector' = 'kafka',
+  'topic' = '%(topic)s',
+  'scan.startup.mode' = 'earliest-offset',
+  'properties.bootstrap.servers' = 'kafka:9094',
+  'format' = 'json'
 );''' % {
           'database': database,
-          'table_name': table_name
+          'table_name': table_name,
+          'columns': ',\n'.join(['  %(name)s %(type)s' % col for col in columns]),
+          'topic': source.get('kafkaSelectedTopics')
       }
 
     if dry_run: