添加日志

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

@@ -96,7 +96,7 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
         if("INSERT".equals(type) || "UPDATE".equals(type)){
             excueteSql = tranferInsertSql(table,dataObj,mysqlType);
         } else {
-            excueteSql = transferDeleteSql(table,dataObj,mysqlType,pkNameSet);
+            excueteSql = logicalDelete ? 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()));