修改错误时间格式的数据

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

@@ -162,6 +162,10 @@ public class MysqlDataTransferFunction extends ProcessFunction<JSONObject, Tuple
             if (d == null) {
                 return "";
             }
+            String originalDateStr = dataObj.getString(columnKey);
+            if (originalDateStr != null && originalDateStr.startsWith("0000-00-00")) {
+                return "'0000-00-00" + ("DATETIME".equals(upperCase) ? " 00:00:00'" : "'");
+            }
             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);

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

@@ -39,7 +39,7 @@ public class MysqlDataSlideSink extends RichSinkFunction<Tuple6<String,String,St
 
     @Override
     public void open(Configuration parameters) throws Exception {
-        executorService = new ThreadPoolExecutor(8, 20, 60,
+        executorService = new ThreadPoolExecutor(5, 10, 60,
                 TimeUnit.SECONDS,
                 new LinkedBlockingDeque<>(QUEUE_CAPACITY),
                 new ThreadPoolExecutor.CallerRunsPolicy());

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

@@ -15,7 +15,7 @@ public class EtlUtils {
         properties.setProperty("auto.offset.reset", "earliest");
         properties.setProperty("sasl.jaas.config", getSaslJaasConfig(username,password));
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
-        properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
+        properties.setProperty("sasl.mechanism", "SCRAM-SHA-256");
         properties.setProperty("fetch.max.bytes", "20971520"); //20M
         properties.setProperty("flink.consumer.max.fetch.size", "20971520");//20M
         properties.setProperty("session.timeout.ms", "60000");