kingbase

pom.xml

@@ -16,6 +16,7 @@
     <fast-json-version>1.2.75</fast-json-version>
     <kafka-clinet-veriosn>2.7.0</kafka-clinet-veriosn>
     <mysql-connector-version>8.0.17</mysql-connector-version>
+    <kingbase8-version>8.6.0</kingbase8-version>
     <flink-jdbc-version>1.10.0</flink-jdbc-version>
     <redisson-version>3.17.7</redisson-version>
     <project.build.sourceEncoding>UTF8</project.build.sourceEncoding>
@@ -351,6 +352,13 @@
       <version>2.1.3</version>
     </dependency>
 
+    <!-- 人大金仓 Kingbase 驱动 -->
+    <dependency>
+      <groupId>com.kingbase8</groupId>
+      <artifactId>kingbase8</artifactId>
+      <version>${kingbase8-version}</version>
+    </dependency>
+
   </dependencies>
 
   <repositories>

src/main/java/com/dsk/flink/dsc/common/function/AsyncMysqlDataTransferFunction.java → src/main/java/com/dsk/flink/dsc/common/function/AsyncDbDataTransferFunction.java

@@ -25,16 +25,16 @@ import java.util.concurrent.TimeUnit;
  * @description mysql/tidb sink 数据组装 sql
  *
  */
-public class AsyncMysqlDataTransferFunction extends RichAsyncFunction<JSONObject,String> {
+public class AsyncDbDataTransferFunction extends RichAsyncFunction<JSONObject,String> {
 
-    static Logger logger = LoggerFactory.getLogger(AsyncMysqlDataTransferFunction.class);
+    static Logger logger = LoggerFactory.getLogger(AsyncDbDataTransferFunction.class);
     //数据库连接信息
     EnvProperties dbInfoMap;
 
     //线程池
     private transient ExecutorService executorService;
 
-    public AsyncMysqlDataTransferFunction(EnvProperties dbInfoMap) {
+    public AsyncDbDataTransferFunction(EnvProperties dbInfoMap) {
         this.dbInfoMap = dbInfoMap;
     }
 
@@ -87,17 +87,23 @@ public class AsyncMysqlDataTransferFunction extends RichAsyncFunction<JSONObject
                     dataObj.put("is_del", "DELETE".equals(type) ? 1 : 0);
                 }
 
+                String dialect = MapUtil.getStr(dbInfoMap, "db_dialect", "mysql");
                 if("INSERT".equals(type)){
-                    excueteSql = tranferInsertSql(table,dataObj,mysqlType);
+                    excueteSql = "kingbase".equalsIgnoreCase(dialect)
+                            ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType);
                 }
                 if("UPDATE".equals(type)){
                     JSONObject oldDataObj = oldDataList.getJSONObject(0);
                     // excueteSql = tranferUpdateSql(table,dataObj,oldDataObj,mysqlType,pkNameSet);
-                    excueteSql = tranferInsertSql(table,dataObj,mysqlType);
+                    excueteSql = "kingbase".equalsIgnoreCase(dialect)
+                            ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType);
                 }
 
                 if("DELETE".equals(type)){
-                    excueteSql = logicalDelete ? tranferInsertSql(table,dataObj,mysqlType) :  transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
+                    excueteSql = logicalDelete
+                            ? ("kingbase".equalsIgnoreCase(dialect)
+                                ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType))
+                            :  transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
                 }
 
                 //处理先后顺序
@@ -151,6 +157,31 @@ public class AsyncMysqlDataTransferFunction extends RichAsyncFunction<JSONObject
         return String.format("INSERT INTO %s (%s) values (%s) ON DUPLICATE KEY UPDATE %s;",table,columnString,valueString,updateString);
     }
 
+    /**
+     * Kingbase（PostgreSQL 系）写法：INSERT ... ON CONFLICT (pk) DO UPDATE
+     */
+    private static String transferUpsertSqlKingbase(String table, JSONObject dataObj, JSONObject mysqlType, Set<String> pkNameSet) {
+        Set<String> columnSet = mysqlType.keySet();
+        StringBuilder colBuilder = new StringBuilder();
+        List<String> valueList = new ArrayList<>();
+        List<String> updateList = new ArrayList<>();
+        for (String s : columnSet) {
+            colBuilder.append(s).append(",");
+            valueList.add(getValueStringKingbase(dataObj, s, mysqlType.getString(s)));
+            updateList.add(s + " = EXCLUDED." + s);
+        }
+        colBuilder.setLength(colBuilder.length() - 1);
+        String valueString = String.join(",", valueList);
+        String updateString = String.join(",", updateList);
+        List<String> pkList = new ArrayList<>(pkNameSet);
+        String pkString = String.join(",", pkList);
+        if (pkList.isEmpty()) {
+            return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);
+        }
+        return String.format("INSERT INTO %s (%s) values (%s) ON CONFLICT (%s) DO UPDATE SET %s",
+                table, colBuilder.toString(), valueString, pkString, updateString);
+    }
+
     private String tranferUpdateSql(String table, JSONObject dataObj, JSONObject oldDataObj, JSONObject mysqlType,Set<String> pkNameSet) {
 
         Set<String> columnSet = mysqlType.keySet();
@@ -208,6 +239,23 @@ public class AsyncMysqlDataTransferFunction extends RichAsyncFunction<JSONObject
         return dataObj.getString(columnKey);
     }
 
+    /**
+     * Kingbase(PostgreSQL 系) 取值：字符串加单引号，内部单引号转义为 ''（而非 \'）
+     */
+    private static String getValueStringKingbase(JSONObject dataObj, String columnKey, String mysqlType) {
+        if (null == dataObj.get(columnKey)) {
+            return "null";
+        }
+        if (Arrays.asList(STR_SQL_TYPE).contains(mysqlType.toUpperCase())) {
+            return "'" + dataObj.getString(columnKey).replace("'", "''") + "'";
+        }
+        if ("DATE".equalsIgnoreCase(mysqlType) || "DATETIME".equalsIgnoreCase(mysqlType)) {
+            SimpleDateFormat df = "DATETIME".equalsIgnoreCase(mysqlType) ? new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") : new SimpleDateFormat("yyyy-MM-dd");
+            return String.format("'%s'", df.format(dataObj.getDate(columnKey)));
+        }
+        return dataObj.getString(columnKey);
+    }
+
 
     public static void main(String[] args) {
         JSONObject jsonObject = new JSONObject();

src/main/java/com/dsk/flink/dsc/common/function/AsyncMysqlDataTransferFunctionNew.java → src/main/java/com/dsk/flink/dsc/common/function/AsyncDbDataTransferFunctionNew.java

@@ -20,16 +20,16 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
-public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObject, Tuple3<String,String,Long>> {
+public class AsyncDbDataTransferFunctionNew extends RichAsyncFunction<JSONObject, Tuple3<String,String,Long>> {
 
-    //static Logger logger = LoggerFactory.getLogger(AsyncMysqlDataTransferFunctionNew.class);
+    //static Logger logger = LoggerFactory.getLogger(AsyncDbDataTransferFunctionNew.class);
     //数据库连接信息
     EnvProperties dbInfoMap;
 
     //线程池
     private transient ExecutorService executorService;
 
-    public AsyncMysqlDataTransferFunctionNew(EnvProperties dbInfoMap) {
+    public AsyncDbDataTransferFunctionNew(EnvProperties dbInfoMap) {
         this.dbInfoMap = dbInfoMap;
     }
 
@@ -94,17 +94,23 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
                     groupKey = table.concat("-").concat(pkValue);
                 }
 
+                String dialect = MapUtil.getStr(dbInfoMap, "db_dialect", "mysql");
                 if("INSERT".equals(type)){
-                    excueteSql = tranferInsertSql(table,dataObj,mysqlType);
+                    excueteSql = "kingbase".equalsIgnoreCase(dialect)
+                            ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType);
                 }
                 if("UPDATE".equals(type)){
                     //JSONObject oldDataObj = oldDataList.getJSONObject(0);
                     //excueteSql = tranferUpdateSql(table,dataObj,oldDataObj,mysqlType,pkNameSet);
-                    excueteSql = tranferInsertSql(table,dataObj,mysqlType);
+                    excueteSql = "kingbase".equalsIgnoreCase(dialect)
+                            ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType);
                 }
 
                 if("DELETE".equals(type)){
-                    excueteSql = logicalDelete ? tranferInsertSql(table,dataObj,mysqlType) :  transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
+                    excueteSql = logicalDelete
+                            ? ("kingbase".equalsIgnoreCase(dialect)
+                                ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet) : tranferInsertSql(table,dataObj,mysqlType))
+                            :  transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
                 }
                 resultList.add(Tuple3.of(excueteSql,groupKey,ts));
                 Boolean logEnable = MapUtil.getBool(dbInfoMap, "log_enable", false);
@@ -122,7 +128,8 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
         });
     }
 
-    private static String logSqlFormat = "INSERT INTO dsc_cdc_log (`table`,op_type,pk_columns,pk_values,data_json,cdc_ts) values ('%s','%s','%s','%s','%s', %d)";
+    private static String logSqlFormatMysql = "INSERT INTO dsc_cdc_log (`table`,op_type,pk_columns,pk_values,data_json,cdc_ts) values ('%s','%s','%s','%s','%s', %d)";
+    private static String logSqlFormatKingbase = "INSERT INTO dsc_cdc_log (table,op_type,pk_columns,pk_values,data_json,cdc_ts) values ('%s','%s','%s','%s','%s', %d)";
 
     private String buildLogData(String type, String table, Set<String> pkNameSet, JSONObject dataObj, long ts, String dataJsonStr) {
         List<String> pkValueList = new ArrayList<>();
@@ -131,8 +138,13 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
         }
         String pkColumns = String.join(",",pkNameSet);
         String pkValues = String.join("-",pkValueList);
-        dataJsonStr = dataJsonStr.replace("\\","\\\\");
-        return String.format(logSqlFormat, table, type, pkColumns, pkValues, dataJsonStr, ts);
+        boolean kingbase = "kingbase".equalsIgnoreCase(MapUtil.getStr(dbInfoMap, "db_dialect", "mysql"));
+        // Kingbase 单引号转义用 ''，MySQL 用 \'
+        dataJsonStr = kingbase
+                ? dataJsonStr.replace("'", "''")
+                : dataJsonStr.replace("\\","\\\\").replace("'", "\\'");
+        String format = kingbase ? logSqlFormatKingbase : logSqlFormatMysql;
+        return String.format(format, table, type, pkColumns, pkValues, dataJsonStr, ts);
     }
 
 
@@ -162,6 +174,31 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
         return String.format("REPLACE INTO %s (%s) values (%s);",table,columnString,valueString);
     }
 
+    /**
+     * Kingbase（PostgreSQL 系）写法：INSERT ... ON CONFLICT (pk) DO UPDATE
+     */
+    private static String transferUpsertSqlKingbase(String table, JSONObject dataObj, JSONObject mysqlType, Set<String> pkNameSet) {
+        Set<String> columnSet = mysqlType.keySet();
+        StringBuilder colBuilder = new StringBuilder();
+        List<String> valueList = new ArrayList<>();
+        List<String> updateList = new ArrayList<>();
+        for (String s : columnSet) {
+            colBuilder.append(s).append(",");
+            valueList.add(getValueStringKingbase(dataObj, s, mysqlType.getString(s)));
+            updateList.add(s + " = EXCLUDED." + s);
+        }
+        colBuilder.setLength(colBuilder.length() - 1);
+        String valueString = String.join(",", valueList);
+        String updateString = String.join(",", updateList);
+        List<String> pkList = new ArrayList<>(pkNameSet);
+        String pkString = String.join(",", pkList);
+        if (pkList.isEmpty()) {
+            return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);
+        }
+        return String.format("INSERT INTO %s (%s) values (%s) ON CONFLICT (%s) DO UPDATE SET %s",
+                table, colBuilder.toString(), valueString, pkString, updateString);
+    }
+
     private String tranferUpdateSql(String table, JSONObject dataObj, JSONObject oldDataObj, JSONObject mysqlType,Set<String> pkNameSet) {
 
         Set<String> columnSet = mysqlType.keySet();
@@ -219,4 +256,21 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
         return dataObj.getString(columnKey);
     }
 
+    /**
+     * Kingbase(PostgreSQL 系) 取值：字符串加单引号，内部单引号转义为 ''（而非 \'）
+     */
+    private static String getValueStringKingbase(JSONObject dataObj, String columnKey, String mysqlType) {
+        if (null == dataObj.get(columnKey)) {
+            return "null";
+        }
+        if (Arrays.asList(STR_SQL_TYPE).contains(mysqlType.toUpperCase())) {
+            return "'" + dataObj.getString(columnKey).replace("'", "''") + "'";
+        }
+        if ("DATE".equalsIgnoreCase(mysqlType) || "DATETIME".equalsIgnoreCase(mysqlType)) {
+            SimpleDateFormat df = "DATETIME".equalsIgnoreCase(mysqlType) ? new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") : new SimpleDateFormat("yyyy-MM-dd");
+            return String.format("'%s'", df.format(dataObj.getDate(columnKey)));
+        }
+        return dataObj.getString(columnKey);
+    }
+
 }

src/main/java/com/dsk/flink/dsc/common/function/MysqlDataTransferFunction.java → src/main/java/com/dsk/flink/dsc/common/function/DbDataTransferFunction.java

@@ -22,7 +22,7 @@ import java.util.*;
  * @author lww
  * @date 2025-01-14
  */
-public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple3<String,String,Long>> {
+public class DbDataTransferFunction extends ProcessFunction<JSONObject, Tuple3<String,String,Long>> {
 
     private static final Map<String,Integer> STR_SQL_TYPE;
     private final EnvProperties dbInfoMap;
@@ -45,7 +45,7 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
         STR_SQL_TYPE.put("JSON",1);
     }
 
-    public MysqlDataTransferFunction(EnvProperties envProps, OutputTag<Tuple6<String,String,String,String,String,Long>> logSlideTag) {
+    public DbDataTransferFunction(EnvProperties envProps, OutputTag<Tuple6<String,String,String,String,String,Long>> logSlideTag) {
         this.dbInfoMap = envProps;
         this.logSlideTag = logSlideTag;
     }
@@ -95,25 +95,35 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
                 table = table.concat("_").concat(String.valueOf(val % i));
             }
         }
+        String dialect = MapUtil.getStr(dbInfoMap, "db_dialect", "mysql");
         if("INSERT".equals(type) || "UPDATE".equals(type)){
-            excueteSql = tranferInsertSql(table,dataObj,mysqlType);
+            excueteSql = "kingbase".equalsIgnoreCase(dialect)
+                    ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet)
+                    : tranferInsertSql(table,dataObj,mysqlType);
         } else {
-            excueteSql = logicalDelete ? tranferInsertSql(table,dataObj,mysqlType) : transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
+            excueteSql = logicalDelete
+                    ? ("kingbase".equalsIgnoreCase(dialect)
+                        ? transferUpsertSqlKingbase(table,dataObj,mysqlType,pkNameSet)
+                        : tranferInsertSql(table,dataObj,mysqlType))
+                    : transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
         }
         if (MapUtil.getBool(dbInfoMap, "log_enable", false)){
-            ctx.output(logSlideTag,buildLogData(type, table, pkColumns, pkColumnVals, ts, value.toJSONString()));
+            ctx.output(logSlideTag,buildLogData(dbInfoMap, type, table, pkColumns, pkColumnVals, ts, value.toJSONString()));
         }
         out.collect(Tuple3.of(excueteSql,groupKey,ts));
     }
 
-    private static Tuple6<String,String,String,String,String,Long> buildLogData(String type, String table, StringBuilder pkColumns, StringBuilder pkValues, long ts, String dataJsonStr) {
+    private static Tuple6<String,String,String,String,String,Long> buildLogData(EnvProperties dbInfoMap, String type, String table, StringBuilder pkColumns, StringBuilder pkValues, long ts, String dataJsonStr) {
         if (pkColumns.length() > 0) {
             pkColumns.setLength(pkColumns.length()-1);
             pkValues.setLength(pkValues.length()-1);
         }
-        String step1 = StrUtil.replace(dataJsonStr, "\\", "\\\\");
-        String step2 = StrUtil.replace(step1, "'", "\\'");
-        return Tuple6.of(table,type, pkColumns.toString(), pkValues.toString().replace("'",""),step2,ts);
+        boolean kingbase = "kingbase".equalsIgnoreCase(MapUtil.getStr(dbInfoMap, "db_dialect", "mysql"));
+        // Kingbase(PostgreSQL) 字符串内的单引号转义为 ''；MySQL 使用 \'
+        String escaped = kingbase
+                ? StrUtil.replace(dataJsonStr, "'", "''")
+                : StrUtil.replace(StrUtil.replace(dataJsonStr, "\\", "\\\\"), "'", "\\'");
+        return Tuple6.of(table,type, pkColumns.toString(), pkValues.toString().replace("'",""),escaped,ts);
     }
 
     private static String tranferInsertSql(String table, JSONObject dataObj, JSONObject mysqlType) {
@@ -131,6 +141,34 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
         return sb.toString();
     }
 
+    /**
+     * Kingbase（PostgreSQL 系）写法：INSERT ... ON CONFLICT (pk) DO UPDATE
+     * 使用双引号包裹标识符（这里为兼容直接去掉特殊符号，采用无引号写法），单引号内转义用 '' 而非 \'
+     */
+    private static String transferUpsertSqlKingbase(String table, JSONObject dataObj, JSONObject mysqlType, Set<String> pkNameSet) {
+        Set<String> columnSet = mysqlType.keySet();
+        StringBuilder colBuilder = new StringBuilder();
+        List<String> valueList = new ArrayList<>();
+        List<String> updateList = new ArrayList<>();
+        for (String s : columnSet) {
+            colBuilder.append(s).append(",");
+            String val = getValueStringKingbase(dataObj, s, mysqlType.getString(s));
+            valueList.add(val);
+            updateList.add(s + " = EXCLUDED." + s);
+        }
+        colBuilder.setLength(colBuilder.length() - 1);
+        String valueString = String.join(",", valueList);
+        String updateString = String.join(",", updateList);
+        List<String> pkList = new ArrayList<>(pkNameSet);
+        String pkString = String.join(",", pkList);
+        // 若没有主键，退化为普通 INSERT（重复会报错，由下游表约束保证）
+        if (pkList.isEmpty()) {
+            return String.format("INSERT INTO %s (%s) values (%s)", table, colBuilder.toString(), valueString);
+        }
+        return String.format("INSERT INTO %s (%s) values (%s) ON CONFLICT (%s) DO UPDATE SET %s",
+                table, colBuilder.toString(), valueString, pkString, updateString);
+    }
+
     private static String transferDeleteSql(String table, JSONObject dataObj, JSONObject mysqlType, Set<String> pkNameSet) {
         StringBuilder whereClauseBuilder = new StringBuilder();
         for (String pk : pkNameSet) {
@@ -169,4 +207,27 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
 
         return dataObj.getString(columnKey);
     }
+
+    /**
+     * Kingbase(PostgreSQL 系) 取值：字符串加单引号，内部单引号转义为 ''（而非 \'）
+     */
+    private static String getValueStringKingbase(JSONObject dataObj, String columnKey, String mysqlType) {
+        if (null == dataObj.get(columnKey)) {
+            return "null";
+        }
+        String upperCase = mysqlType.toUpperCase();
+        if (STR_SQL_TYPE.containsKey(upperCase)) {
+            return "'" + StrUtil.replace(dataObj.getString(columnKey), "'", "''") + "'";
+        }
+        if ("DATE".equals(upperCase) || "DATETIME".equals(upperCase)) {
+            Date d = dataObj.getDate(columnKey);
+            if (d == null) {
+                return "";
+            }
+            LocalDateTime dateTime = LocalDateTime.ofInstant(d.toInstant(), ZoneId.systemDefault());
+            String date = "DATETIME".equals(upperCase) ? DATETIME_FORMAT.format(dateTime) : DATE_FORMAT.format(dateTime);
+            return String.format("'%s'", date);
+        }
+        return dataObj.getString(columnKey);
+    }
 }

src/main/java/com/dsk/flink/dsc/common/sink/MysqlDataSlideSink.java → src/main/java/com/dsk/flink/dsc/common/sink/DbDataSlideSink.java

@@ -3,6 +3,7 @@ package com.dsk.flink.dsc.common.sink;
 import cn.hutool.core.lang.Snowflake;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.RandomUtil;
+import cn.hutool.core.util.StrUtil;
 import cn.hutool.db.DbUtil;
 import cn.hutool.db.sql.SqlExecutor;
 import com.alibaba.druid.pool.DruidDataSource;
@@ -22,17 +23,17 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
-public class MysqlDataSlideSink extends RichSinkFunction<Tuple6<String,String,String,String,String,Long>> {
+public class DbDataSlideSink extends RichSinkFunction<Tuple6<String,String,String,String,String,Long>> {
 
-    static Logger logger = LoggerFactory.getLogger(MysqlDataSlideSink.class);
+    static Logger logger = LoggerFactory.getLogger(DbDataSlideSink.class);
     EnvProperties envProps;
     private transient ExecutorService executorService;
     private transient DruidDataSource dataSource;
     private static final int MAX_RETRIES = 3; // 最大重试次数
     private static final int RETRY_DELAY_MS = 100; // 重试间隔时间
-    private static final String SQL = "INSERT INTO dsc_cdc_log (`table`,op_type,pk_columns,pk_values,data_json,cdc_ts) values (?,?,?,?,?,?)";
+    private static final String SQL = "INSERT INTO dsc_cdc_log (table,op_type,pk_columns,pk_values,data_json,cdc_ts) values (?,?,?,?,?,?)";
 
-    public MysqlDataSlideSink(EnvProperties envProps) {
+    public DbDataSlideSink(EnvProperties envProps) {
         this.envProps = envProps;
     }
 
@@ -43,7 +44,13 @@ public class MysqlDataSlideSink extends RichSinkFunction<Tuple6<String,String,St
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         //System.out.println(configTidbUrl);
         dataSource = new DruidDataSource();
-        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        String driverClass = envProps.getDb_driver();
+        if (StrUtil.isBlank(driverClass)) {
+            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+        }
+        dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
         dataSource.setPassword(envProps.getDb_password());
         dataSource.setUrl(configTidbUrl);
@@ -100,7 +107,7 @@ public class MysqlDataSlideSink extends RichSinkFunction<Tuple6<String,String,St
 
     private void writeErrLogDb(SqlErrorLog errorLog) {
         Snowflake snowflake = IdUtil.getSnowflake(RandomUtil.randomInt(31), RandomUtil.randomInt(31));
-        String sql = "insert dsc_err_log (id,error_time, error_sql, error_msg) values (?, ?, ?, ?)";
+        String sql = "insert into dsc_err_log (id,error_time, error_sql, error_msg) values (?, ?, ?, ?)";
         Connection conn = null;
         PreparedStatement pt = null;
         try {

src/main/java/com/dsk/flink/dsc/common/sink/MysqlDataTransferSink.java → src/main/java/com/dsk/flink/dsc/common/sink/DbDataTransferSink.java

@@ -3,6 +3,7 @@ package com.dsk.flink.dsc.common.sink;
 import cn.hutool.core.lang.Snowflake;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.RandomUtil;
+import cn.hutool.core.util.StrUtil;
 import cn.hutool.db.DbUtil;
 import cn.hutool.db.sql.SqlExecutor;
 import com.alibaba.druid.pool.DruidDataSource;
@@ -21,16 +22,16 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
-public class MysqlDataTransferSink extends RichSinkFunction<String> {
+public class DbDataTransferSink extends RichSinkFunction<String> {
 
-    static Logger logger = LoggerFactory.getLogger(MysqlDataTransferSink.class);
+    static Logger logger = LoggerFactory.getLogger(DbDataTransferSink.class);
     EnvProperties envProps;
     private transient ExecutorService executorService;
     private transient DruidDataSource dataSource;
     private static final int MAX_RETRIES = 3; // 最大重试次数
     private static final int RETRY_DELAY_MS = 100; // 重试间隔时间
 
-    public MysqlDataTransferSink(EnvProperties envProps) {
+    public DbDataTransferSink(EnvProperties envProps) {
         this.envProps = envProps;
     }
 
@@ -41,7 +42,13 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         //System.out.println(configTidbUrl);
         dataSource = new DruidDataSource();
-        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        String driverClass = envProps.getDb_driver();
+        if (StrUtil.isBlank(driverClass)) {
+            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+        }
+        dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
         dataSource.setPassword(envProps.getDb_password());
         dataSource.setUrl(configTidbUrl);
@@ -117,7 +124,7 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
 
     private void writeErrLogDb(SqlErrorLog errorLog) {
         Snowflake snowflake = IdUtil.getSnowflake(RandomUtil.randomInt(31), RandomUtil.randomInt(31));
-        String sql = "insert dsc_err_log (id,error_time, error_sql, error_msg) values (?, ?, ?, ?)";
+        String sql = "insert into dsc_err_log (id,error_time, error_sql, error_msg) values (?, ?, ?, ?)";
         Connection conn = null;
         PreparedStatement pt = null;
         try {

src/main/java/com/dsk/flink/dsc/common/sink/MysqlDataTransferSinkBatch.java → src/main/java/com/dsk/flink/dsc/common/sink/DbDataTransferSinkBatch.java

@@ -2,6 +2,7 @@ package com.dsk.flink.dsc.common.sink;
 
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.lang.Snowflake;
+import cn.hutool.core.util.StrUtil;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.RandomUtil;
 import com.alibaba.druid.pool.DruidDataSource;
@@ -25,9 +26,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * @author lww
  * @date 2025-01-14
  */
-public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
+public class DbDataTransferSinkBatch extends RichSinkFunction<String> {
 
-    static Logger logger = LoggerFactory.getLogger(MysqlDataTransferSink.class);
+    static Logger logger = LoggerFactory.getLogger(DbDataTransferSinkBatch.class);
     EnvProperties envProps;
     private transient ExecutorService executorService;
     private transient DruidDataSource dataSource;
@@ -40,7 +41,7 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
     private AtomicBoolean flushing = new AtomicBoolean(false);
     private int subtaskIndex;
 
-    public MysqlDataTransferSinkBatch(EnvProperties envProps) {
+    public DbDataTransferSinkBatch(EnvProperties envProps) {
         this.envProps = envProps;
     }
 
@@ -51,7 +52,13 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
         // 初始化获取配置
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         dataSource = new DruidDataSource();
-        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
+        // 驱动类名优先取配置 db_driver，未配置则按方言推断，仍无则默认 MySQL
+        String driverClass = envProps.getDb_driver();
+        if (StrUtil.isBlank(driverClass)) {
+            driverClass = "kingbase".equalsIgnoreCase(envProps.getDb_dialect())
+                    ? "com.kingbase8.Driver" : "com.mysql.cj.jdbc.Driver";
+        }
+        dataSource.setDriverClassName(driverClass);
         dataSource.setUsername(envProps.getDb_username());
         dataSource.setPassword(envProps.getDb_password());
         dataSource.setUrl(configTidbUrl);
@@ -60,7 +67,10 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
         dataSource.setTestWhileIdle(true);
         dataSource.setMaxWait(20000);
         dataSource.setValidationQuery("select 1");
-        dataSource.setConnectionInitSqls(CollUtil.newArrayList("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"));
+        // MySQL 专属的连接初始化 SQL，Kingbase 不兼容，按方言跳过
+        if (!"kingbase".equalsIgnoreCase(envProps.getDb_dialect())) {
+            dataSource.setConnectionInitSqls(CollUtil.newArrayList("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"));
+        }
         scheduledExecutorService = new ScheduledThreadPoolExecutor(1);
         scheduledExecutorService.scheduleAtFixedRate(this::flush, FLUSH_INTERVAL, FLUSH_INTERVAL, TimeUnit.MILLISECONDS);
         logger.info("Subtask {} initialized with batch size {}", subtaskIndex, BATCH_SIZE);

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -3,9 +3,9 @@ package com.dsk.flink.dsc.sync;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.StrUtil;
 import com.alibaba.fastjson.JSONObject;
-import com.dsk.flink.dsc.common.function.MysqlDataTransferFunction;
-import com.dsk.flink.dsc.common.sink.MysqlDataSlideSink;
-import com.dsk.flink.dsc.common.sink.MysqlDataTransferSink;
+import com.dsk.flink.dsc.common.function.DbDataTransferFunction;
+import com.dsk.flink.dsc.common.sink.DbDataSlideSink;
+import com.dsk.flink.dsc.common.sink.DbDataTransferSink;
 import com.dsk.flink.dsc.utils.EnvProperties;
 import com.dsk.flink.dsc.utils.EnvPropertiesUtil;
 import com.dsk.flink.dsc.utils.EtlUtils;
@@ -108,7 +108,7 @@ public class SyncCustomerDataSource {
         OutputTag<Tuple6<String,String,String,String,String,Long>> logSlideTag = new OutputTag<Tuple6<String,String,String,String,String,Long>>("log_slide") {};
 
         SingleOutputStreamOperator<Tuple3<String, String, Long>> slide = tsGroupStream
-                .process(new MysqlDataTransferFunction(envProps,logSlideTag))
+                .process(new DbDataTransferFunction(envProps,logSlideTag))
                 .name("dsc-sql")
                 .uid("dsc-sql");
 
@@ -134,7 +134,7 @@ public class SyncCustomerDataSource {
                 .name("dsc-max")
                 .uid("dsc-max");
 
-        groupWindowSqlResultStream.addSink(new MysqlDataTransferSink(envProps))
+        groupWindowSqlResultStream.addSink(new DbDataTransferSink(envProps))
                 .name("dsc-sink")
                 .uid("dsc-sink");
 
@@ -164,7 +164,7 @@ public class SyncCustomerDataSource {
         )).uid("dsc-log")
           .name("dsc-log");*/
 
-        sideOutput.addSink(new MysqlDataSlideSink(envProps)).uid("dsc-log")
+        sideOutput.addSink(new DbDataSlideSink(envProps)).uid("dsc-log")
                 .name("dsc-log");
 
         env.execute("dsc-client");

src/main/java/com/dsk/flink/dsc/utils/EnvProperties.java

@@ -13,6 +13,12 @@ public class EnvProperties extends Properties {
 	// #连接
 	String db_url;
 
+	// #数据库驱动类全名（如 com.mysql.cj.jdbc.Driver / com.kingbase8.Driver）
+	String db_driver;
+
+	// #数据库方言：mysql / kingbase，决定 SQL 生成与转义规则
+	String db_dialect;
+
 	// #建设库TIDB库
 	String db_host;
 	String db_port;
@@ -555,4 +561,20 @@ public class EnvProperties extends Properties {
 	public void setKafka_topic(String kafka_topic) {
 		this.kafka_topic = kafka_topic;
 	}
+
+	public String getDb_driver() {
+		return db_driver == null ? this.getProperty("db_driver") : db_driver;
+	}
+
+	public void setDb_driver(String db_driver) {
+		this.db_driver = db_driver;
+	}
+
+	public String getDb_dialect() {
+		return db_dialect == null ? this.getProperty("db_dialect") : db_dialect;
+	}
+
+	public void setDb_dialect(String db_dialect) {
+		this.db_dialect = db_dialect;
+	}
 }