数据源元数据生成Flink DDL

dlink-admin/src/main/java/com/dlink/controller/DataBaseController.java

@@ -145,4 +145,12 @@ public class DataBaseController {
     public Result listColumns(@RequestParam Integer id,@RequestParam String schemaName,@RequestParam String tableName) {
         return Result.succeed(databaseService.listColumns(id,schemaName,tableName),"获取成功");
     }
+
+    /**
+     * 获取 FlinkTableSql
+     */
+    @GetMapping("/getFlinkTableSql")
+    public Result getFlinkTableSql(@RequestParam Integer id,@RequestParam String schemaName,@RequestParam String tableName) {
+        return Result.succeed(databaseService.getFlinkTableSql(id,schemaName,tableName),"获取成功");
+    }
 }
\ No newline at end of file

dlink-admin/src/main/java/com/dlink/model/DataBase.java

@@ -38,6 +38,8 @@ public class DataBase extends SuperEntity {
 
     private String note;
 
+    private String flinkConfig;
+
     private String dbVersion;
 
     private boolean status;

dlink-admin/src/main/java/com/dlink/service/DataBaseService.java

@@ -26,4 +26,6 @@ public interface DataBaseService extends ISuperService<DataBase> {
     List<Schema> getSchemasAndTables(Integer id);
 
     List<Column> listColumns(Integer id, String schemaName, String tableName);
+
+    String getFlinkTableSql(Integer id, String schemaName, String tableName);
 }

dlink-admin/src/main/java/com/dlink/service/impl/DataBaseServiceImpl.java

@@ -10,10 +10,12 @@ import com.dlink.metadata.driver.DriverConfig;
 import com.dlink.model.Column;
 import com.dlink.model.DataBase;
 import com.dlink.model.Schema;
+import com.dlink.model.Table;
 import com.dlink.service.DataBaseService;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
+import java.util.HashMap;
 import java.util.List;
 
 
@@ -89,4 +91,14 @@ public class DataBaseServiceImpl extends SuperServiceImpl<DataBaseMapper, DataBa
         driver.close();
         return columns;
     }
+
+    @Override
+    public String getFlinkTableSql(Integer id, String schemaName, String tableName) {
+        DataBase dataBase = getById(id);
+        Asserts.checkNotNull(dataBase,"该数据源不存在！");
+        Driver driver = Driver.build(dataBase.getDriverConfig()).connect();
+        List<Column> columns = driver.listColumns(schemaName, tableName);
+        Table table = Table.build(tableName, schemaName, columns);
+        return table.getFlinkTableSql(dataBase.getName(),driver.getFlinkColumnTypeConversion(),dataBase.getFlinkConfig());
+    }
 }

dlink-admin/src/main/resources/mapper/DataBaseMapper.xml

@@ -14,6 +14,7 @@
         <result column="password" property="password" />
         <result column="db_version" property="dbVersion" />
         <result column="note" property="note" />
+        <result column="flink_config" property="flinkConfig" />
         <result column="status" property="status" />
         <result column="health_time" property="healthTime" />
         <result column="heartbeat_time" property="heartbeatTime" />
@@ -24,7 +25,7 @@
 
     <!-- 通用查询结果列 -->
     <sql id="Base_Column_List">
-        id, name, alias,group_name, type,url,username,password,db_version,note,status,health_time,heartbeat_time, enabled, create_time, update_time
+        id, name, alias,group_name, type,url,username,password,db_version,note,flink_config,status,health_time,heartbeat_time, enabled, create_time, update_time
     </sql>
 
 

dlink-common/src/main/java/com/dlink/model/Table.java

 package com.dlink.model;
 
+import com.dlink.assertion.Asserts;
+import com.dlink.utils.SqlUtil;
 import lombok.Getter;
 import lombok.Setter;
 
 import java.io.Serializable;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Table
@@ -57,4 +61,47 @@ public class Table implements Serializable, Comparable<Table> {
     public static Table build(String name, String schema, List<Column> columns) {
         return new Table(name, schema, columns);
     }
+
+    public String getFlinkTableWith(String flinkConfig) {
+        String tableWithSql = "";
+        if (Asserts.isNotNullString(flinkConfig)) {
+            tableWithSql = SqlUtil.replaceAllParam(flinkConfig, "schemaName", schema);
+            tableWithSql = SqlUtil.replaceAllParam(tableWithSql, "tableName", name);
+        }
+        return tableWithSql;
+    }
+
+    public String getFlinkTableSql(String catalogName, Map<String, String> typeConversion,String flinkConfig) {
+        StringBuilder sb = new StringBuilder("CREATE TABLE ");
+        sb.append(catalogName + "." + schema + "." + name + " (\n");
+        List<String> pks = new ArrayList<>();
+        for (int i = 0; i < columns.size(); i++) {
+            String type = typeConversion.containsKey(columns.get(i).getType()) ?
+                    typeConversion.get(columns.get(i).getType()) : "STRING";
+            sb.append("    ");
+            if (i > 0) {
+                sb.append(",");
+            }
+            sb.append(columns.get(i).getName() + " " + type + "\n");
+            if (columns.get(i).isKeyFlag()) {
+                pks.add(columns.get(i).getName());
+            }
+        }
+        StringBuilder pksb = new StringBuilder("PRIMARY KEY ( ");
+        for (int i = 0; i < pks.size(); i++) {
+            if (i > 0) {
+                pksb.append(",");
+            }
+            pksb.append(pks.get(i));
+        }
+        pksb.append(" ) NOT ENFORCED\n");
+        if (pks.size() > 0) {
+            sb.append("    ,");
+            sb.append(pksb);
+        }
+        sb.append(") WITH (\n");
+        sb.append(getFlinkTableWith(flinkConfig));
+        sb.append(");\n");
+        return sb.toString();
+    }
 }

dlink-common/src/main/java/com/dlink/utils/SqlUtil.java

@@ -10,17 +10,21 @@ import com.dlink.assertion.Asserts;
  */
 public class SqlUtil {
 
-    public static String[] getStatements(String sql,String sqlSeparator){
-        if(Asserts.isNullString(sql)){
+    public static String[] getStatements(String sql, String sqlSeparator) {
+        if (Asserts.isNullString(sql)) {
             return new String[0];
         }
         return sql.split(sqlSeparator);
     }
 
-    public static String removeNote(String sql){
-        if(Asserts.isNotNullString(sql)) {
+    public static String removeNote(String sql) {
+        if (Asserts.isNotNullString(sql)) {
             sql = sql.replaceAll("\u00A0", " ").replaceAll("--([^'\r\n]{0,}('[^'\r\n]{0,}'){0,1}[^'\r\n]{0,}){0,}", "").replaceAll("[\r\n]+", "\r\n").trim();
         }
         return sql;
     }
+
+    public static String replaceAllParam(String sql, String name, String value) {
+        return sql.replaceAll("\\$\\{"+name+"\\}",value);
+    }
 }

dlink-doc/sql/dlink.sql

@@ -93,6 +93,7 @@ CREATE TABLE `dlink_database`  (
 	`username` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '用户名',
 	`password` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '密码',
 	`note` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '注释',
+	`flink_config` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT 'Flink配置',
 	`db_version` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '版本，如oracle的11g，hbase的2.2.3',
 	`status` tinyint(1) NULL DEFAULT 0 COMMENT '状态',
 	`health_time` datetime(0) NULL DEFAULT NULL COMMENT '最近健康时间',

dlink-doc/sql/dlink_history.sql

@@ -485,4 +485,10 @@ ADD COLUMN `database_id` int(11) NULL COMMENT '数据源ID' AFTER `cluster_confi
 ALTER TABLE `dlink_task`
 ADD COLUMN `env_id` int(11) NULL COMMENT '环境ID' AFTER `jar_id`;
 
+-- ----------------------------
+-- 0.6.0-SNAPSHOT 2022-01-28
+-- ----------------------------
+ALTER TABLE `dlink_database`
+ADD COLUMN `flink_config` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL COMMENT '环境ID' AFTER `note`;
+
 SET FOREIGN_KEY_CHECKS = 1;

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/AbstractDriver.java

@@ -46,4 +46,5 @@ public abstract class AbstractDriver implements Driver {
     public boolean existTable(Table table){
         return listTables(table.getSchema()).stream().anyMatch(tableItem -> Asserts.isEquals(tableItem.getName(),table.getName()));
     }
+
 }
\ No newline at end of file

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/AbstractJdbcDriver.java

@@ -23,10 +23,7 @@ import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
+import java.util.*;
 
 /**
  * AbstractJdbcDriver
@@ -362,6 +359,10 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
         } finally {
             return sqlExplainResults;
         }
+    }
 
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
     }
 }

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/Driver.java

@@ -12,6 +12,7 @@ import sun.misc.Service;
 import java.sql.SQLException;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 /**
@@ -98,4 +99,5 @@ public interface Driver {
 
     List<SqlExplainResult> explain(String sql);
 
+    Map<String,String> getFlinkColumnTypeConversion();
 }

dlink-metadata/dlink-metadata-clickhouse/src/main/java/com/dlink/metadata/driver/ClickHouseDriver.java

@@ -19,9 +19,7 @@ import com.dlink.result.SqlExplainResult;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Locale;
+import java.util.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -163,4 +161,9 @@ public class ClickHouseDriver extends AbstractJdbcDriver {
             return SqlExplainResult.success(sqlStatement.getClass().getSimpleName(), sqlStatement.toString(), null);
         }
     }
+
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
+    }
 }

dlink-metadata/dlink-metadata-doris/src/main/java/com/dlink/metadata/driver/DorisDriver.java

@@ -9,7 +9,9 @@ import com.dlink.model.Column;
 import com.dlink.model.Table;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 public class DorisDriver extends  AbstractJdbcDriver{
     @Override
@@ -80,4 +82,9 @@ public class DorisDriver extends  AbstractJdbcDriver{
         sb.append("PROPERTIES('replication_num' = '1')");
         return sb.toString();
     }
+
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
+    }
 }

dlink-metadata/dlink-metadata-mysql/src/main/java/com/dlink/metadata/driver/MySqlDriver.java

@@ -9,7 +9,9 @@ import com.dlink.model.Column;
 import com.dlink.model.Table;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * MysqlDriver
@@ -83,4 +85,13 @@ public class MySqlDriver extends AbstractJdbcDriver {
         return sb.toString();
     }
 
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        HashMap<String,String> map = new HashMap<>();
+        map.put("varchar","STRING");
+        map.put("text","STRING");
+        map.put("int","INT");
+        map.put("datetime","TIMESTAMP");
+        return map;
+    }
 }

dlink-metadata/dlink-metadata-oracle/src/main/java/com/dlink/metadata/driver/OracleDriver.java

@@ -8,7 +8,9 @@ import com.dlink.metadata.query.OracleQuery;
 import com.dlink.model.Column;
 import com.dlink.model.Table;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 /**
@@ -78,4 +80,8 @@ public class OracleDriver extends AbstractJdbcDriver {
         return sb.toString();
     }
 
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
+    }
 }

dlink-metadata/dlink-metadata-postgresql/src/main/java/com/dlink/metadata/driver/PostgreSqlDriver.java

@@ -6,6 +6,9 @@ import com.dlink.metadata.query.IDBQuery;
 import com.dlink.metadata.query.PostgreSqlQuery;
 import com.dlink.model.Table;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * PostgreSqlDriver
  *
@@ -42,4 +45,9 @@ public class PostgreSqlDriver extends AbstractJdbcDriver {
     public String getCreateTableSql(Table table) {
         return null;
     }
+
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
+    }
 }

dlink-metadata/dlink-metadata-sqlserver/src/main/java/com/dlink/metadata/driver/SqlServerDriver.java

@@ -8,7 +8,9 @@ import com.dlink.metadata.query.SqlServerQuery;
 import com.dlink.model.Column;
 import com.dlink.model.Table;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  *
@@ -85,4 +87,9 @@ public class SqlServerDriver  extends AbstractJdbcDriver {
         }
         return sb.toString();
     }
+
+    @Override
+    public Map<String,String> getFlinkColumnTypeConversion(){
+        return new HashMap<>();
+    }
 }

dlink-web/src/components/Studio/StudioEvent/Utils.ts

@@ -35,7 +35,7 @@ export function parseSqlMetaData(sql:string) {
  }
 
  function regMetaData(statement:string){
-   const regTable = new RegExp(/create\s+table\s+(\w+?)\s*\(/, 'ig');
+   const regTable = new RegExp(/create\s+table\s+(.+?)\s*\(/, 'ig');
    let table = RegStr(statement,regTable);
    const regConnector = new RegExp(/connector'\s*=\s*'(\w+?)'/, 'ig');
    let connector = RegStr(statement,regConnector);

dlink-web/src/pages/DataBase/components/ClickHouseForm.tsx

@@ -31,6 +31,7 @@ const ClickHouseForm: React.FC<ClickHouseFormProps> = (props) => {
     username: props.values.username,
     password: props.values.password,
     dbVersion: props.values.dbVersion,
+    flinkConfig: props.values.flinkConfig,
     note: props.values.note,
     enabled: props.values.enabled,
   });
@@ -106,6 +107,13 @@ const ClickHouseForm: React.FC<ClickHouseFormProps> = (props) => {
         >
           <Input.Password/>
         </Form.Item>
+        <Form.Item
+          name="flinkConfig"
+          label="Flink With 模板"
+        >
+          <TextArea placeholder="'connector' = 'jdbc',..." allowClear
+                    autoSize={{minRows: 3, maxRows: 10}}/>
+        </Form.Item>
         <Form.Item
           name="note"
           label="注释"
@@ -140,6 +148,7 @@ const ClickHouseForm: React.FC<ClickHouseFormProps> = (props) => {
                 username: formVals.username,
                 password: formVals.password,
                 note: formVals.note,
+                flinkConfig: formVals.flinkConfig,
                 enabled: formVals.enabled,
               }}
             >

dlink-web/src/pages/DataBase/components/DorisForm.tsx

@@ -31,6 +31,7 @@ const DorisForm: React.FC<DorisFormProps> = (props) => {
     username: props.values.username,
     password: props.values.password,
     dbVersion: props.values.dbVersion,
+    flinkConfig: props.values.flinkConfig,
     note: props.values.note,
     enabled: props.values.enabled,
   });
@@ -106,6 +107,13 @@ const DorisForm: React.FC<DorisFormProps> = (props) => {
         >
           <Input.Password/>
         </Form.Item>
+        <Form.Item
+          name="flinkConfig"
+          label="Flink With 模板"
+        >
+          <TextArea placeholder="'connector' = 'doris',..." allowClear
+                    autoSize={{minRows: 3, maxRows: 10}}/>
+        </Form.Item>
         <Form.Item
           name="note"
           label="注释"
@@ -140,6 +148,7 @@ const DorisForm: React.FC<DorisFormProps> = (props) => {
                 username: formVals.username,
                 password: formVals.password,
                 note: formVals.note,
+                flinkConfig: formVals.flinkConfig,
                 enabled: formVals.enabled,
               }}
             >

dlink-web/src/pages/DataBase/components/MySqlForm.tsx

@@ -31,6 +31,7 @@ const MysqlForm: React.FC<MysqlFormProps> = (props) => {
     username: props.values.username,
     password: props.values.password,
     dbVersion: props.values.dbVersion,
+    flinkConfig: props.values.flinkConfig,
     note: props.values.note,
     enabled: props.values.enabled,
   });
@@ -106,6 +107,13 @@ const MysqlForm: React.FC<MysqlFormProps> = (props) => {
         >
           <Input.Password/>
         </Form.Item>
+        <Form.Item
+          name="flinkConfig"
+          label="Flink With 模板"
+        >
+          <TextArea placeholder="'connector' = 'jdbc',..." allowClear
+                    autoSize={{minRows: 3, maxRows: 10}}/>
+        </Form.Item>
         <Form.Item
           name="note"
           label="注释"
@@ -140,6 +148,7 @@ const MysqlForm: React.FC<MysqlFormProps> = (props) => {
                 username: formVals.username,
                 password: formVals.password,
                 note: formVals.note,
+                flinkConfig: formVals.flinkConfig,
                 enabled: formVals.enabled,
               }}
             >

dlink-web/src/pages/DataBase/components/OracleForm.tsx

@@ -31,6 +31,7 @@ const OracleForm: React.FC<OracleFormProps> = (props) => {
     username: props.values.username,
     password: props.values.password,
     dbVersion: props.values.dbVersion,
+    flinkConfig: props.values.flinkConfig,
     note: props.values.note,
     enabled: props.values.enabled,
   });
@@ -106,6 +107,13 @@ const OracleForm: React.FC<OracleFormProps> = (props) => {
         >
           <Input.Password/>
         </Form.Item>
+        <Form.Item
+          name="flinkConfig"
+          label="Flink With 模板"
+        >
+          <TextArea placeholder="'connector' = 'jdbc',..." allowClear
+                    autoSize={{minRows: 3, maxRows: 10}}/>
+        </Form.Item>
         <Form.Item
           name="note"
           label="注释"
@@ -140,6 +148,7 @@ const OracleForm: React.FC<OracleFormProps> = (props) => {
                 username: formVals.username,
                 password: formVals.password,
                 note: formVals.note,
+                flinkConfig: formVals.flinkConfig,
                 enabled: formVals.enabled,
               }}
             >

dlink-web/src/pages/DataBase/components/SqlServerForm.tsx

@@ -31,6 +31,7 @@ const SqlServerForm: React.FC<SqlServerFormProps> = (props) => {
     username: props.values.username,
     password: props.values.password,
     dbVersion: props.values.dbVersion,
+    flinkConfig: props.values.flinkConfig,
     note: props.values.note,
     enabled: props.values.enabled,
   });
@@ -106,6 +107,13 @@ const SqlServerForm: React.FC<SqlServerFormProps> = (props) => {
         >
           <Input.Password/>
         </Form.Item>
+        <Form.Item
+          name="flinkConfig"
+          label="Flink With 模板"
+        >
+          <TextArea placeholder="'connector' = 'jdbc',..." allowClear
+                    autoSize={{minRows: 3, maxRows: 10}}/>
+        </Form.Item>
         <Form.Item
           name="note"
           label="注释"
@@ -140,6 +148,7 @@ const SqlServerForm: React.FC<SqlServerFormProps> = (props) => {
                 username: formVals.username,
                 password: formVals.password,
                 note: formVals.note,
+                flinkConfig: formVals.flinkConfig,
                 enabled: formVals.enabled,
               }}
             >

dlink-web/src/pages/DataBase/data.d.ts

@@ -8,6 +8,7 @@ export type DataBaseItem = {
   username: string,
   password: string,
   note: string,
+  flinkConfig: string,
   dbVersion: string,
   status: boolean,
   healthTime: Date,
@@ -28,6 +29,7 @@ export type DataBaseFormProps = {
   username: string,
   password: string,
   note: string,
+  flinkConfig: string,
   dbVersion: string,
   enabled: boolean,
 }