update codes

src/main/java/com/gmei/data/monitor/PortraitMonitorMain.java

 package com.gmei.data.monitor;
 
-import com.gmei.data.monitor.operator.PortraitMonitorErrOperator;
 import com.gmei.data.monitor.operator.PortraitMonitorShdOperator;
 import com.gmei.data.monitor.operator.PortraitMonitorSucOperator;
 import com.gmei.data.monitor.source.PortraitKafkaSource;
@@ -13,8 +12,6 @@ import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
-import java.text.SimpleDateFormat;
-
 /**
  * @ClassName PortraitMonitorMain
  * @Description: 画像打点实时监控主入口
@@ -33,7 +30,6 @@ public class PortraitMonitorMain {
         String maidianInTopic = parameterTool.get("maidianInTopic", "test11");
         String backendInTopic = parameterTool.get("backendInTopic","test12");
         String portraitSucInTopic = parameterTool.get("portraitSucInTopic","test13");
-        String portraitErrGroupId = parameterTool.get("portraitErrGroupId","flink_monitor_err");
         String portraitShdGroupId = parameterTool.get("portraitShdGroupId","flink_monitor_shd");
         String portraitSucGroupId = parameterTool.get("portraitSucGroupId","flink_monitor_suc");
         Integer windowSize = parameterTool.getInt("windowSize",60);
@@ -57,18 +53,6 @@ public class PortraitMonitorMain {
         CheckpointConfig config = env.getCheckpointConfig();
         config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
-        // 获取数据源
-        DataStream portraitErrDataStream = new PortraitKafkaSource(
-                env,
-                inBrokers,
-                maidianInTopic,
-                backendInTopic,
-                portraitErrGroupId,
-                batchSize,
-                isStartFromEarliest,
-                isStartFromLatest,
-                startTime
-        ).getInstance();
         DataStream portraitShdDataStream = new PortraitKafkaSource(
                 env,
                 inBrokers,
@@ -92,7 +76,6 @@ public class PortraitMonitorMain {
         ).getInstance();
 
         // 执行处理核心逻辑
-        new PortraitMonitorErrOperator(portraitErrDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
         new PortraitMonitorShdOperator(portraitShdDataStream,windowSize,slideSize,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
         new PortraitMonitorSucOperator(portraitSucDataStream,windowSize,slideSize,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
 

src/main/java/com/gmei/data/monitor/PortraitMonitorMainShdSuc.java → src/main/java/com/gmei/data/monitor/PortraitMonitorMainAll.java

 package com.gmei.data.monitor;
 
+import com.gmei.data.monitor.operator.PortraitMonitorErrOperator;
 import com.gmei.data.monitor.operator.PortraitMonitorShdOperator;
 import com.gmei.data.monitor.operator.PortraitMonitorSucOperator;
 import com.gmei.data.monitor.source.PortraitKafkaSource;
@@ -20,7 +21,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  * @Date 2020/3/18
  * @Version V1.0
  **/
-public class PortraitMonitorMainShdSuc {
+public class PortraitMonitorMainAll {
 
     public static void main(String[] args) throws Exception{
         // 获取运行参数
@@ -55,17 +56,17 @@ public class PortraitMonitorMainShdSuc {
         config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
         // 获取数据源
-        //        DataStream portraitErrDataStream = new PortraitKafkaSource(
-        //                env,
-        //                inBrokers,
-        //                maidianInTopic,
-        //                backendInTopic,
-        //                portraitErrGroupId,
-        //                batchSize,
-        //                isStartFromEarliest,
-        //                isStartFromLatest,
-        //                startTime
-        //        ).getInstance();
+        DataStream portraitErrDataStream = new PortraitKafkaSource(
+                env,
+                inBrokers,
+                maidianInTopic,
+                backendInTopic,
+                portraitErrGroupId,
+                batchSize,
+                isStartFromEarliest,
+                isStartFromLatest,
+                startTime
+        ).getInstance();
         DataStream portraitShdDataStream = new PortraitKafkaSource(
                 env,
                 inBrokers,
@@ -89,7 +90,7 @@ public class PortraitMonitorMainShdSuc {
         ).getInstance();
 
         // 执行处理核心逻辑
-        // new PortraitMonitorErrOperator(portraitErrDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
+        new PortraitMonitorErrOperator(portraitErrDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
         new PortraitMonitorShdOperator(portraitShdDataStream,windowSize,slideSize,outJdbcUrl,maxRetry,retryInteral,parallelism).run();
         new PortraitMonitorSucOperator(portraitSucDataStream,windowSize,slideSize,outJdbcUrl,maxRetry,retryInteral,parallelism).run();