修改kafka连接

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

@@ -152,7 +152,7 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
         //需要处理成字符串加引号的类型
         if(STR_SQL_TYPE.containsKey(upperCase)){
             String step1 = StrUtil.replace(dataObj.getString(columnKey), "\\", "\\\\");
-            return StrUtil.replace(step1, "'", "\\'");
+            return "'" + StrUtil.replace(step1, "'", "\\'") + "'";
             //return String.format("'%s'", dataObj.getString(columnKey).replace("\\","\\\\").replace("'", "\\'") );
         }
 
@@ -164,7 +164,7 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
             }
             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 String.format("'%s'",date);
         }
 
         return dataObj.getString(columnKey);