update codes

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

@@ -49,11 +49,11 @@ public class PortraitMonitorMain {
 
         // 获得流处理环境对象
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-//        env.enableCheckpointing(1000);
-//        env.setStateBackend(new FsStateBackend(checkpointPath));
-//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 3000));
-//        CheckpointConfig config = env.getCheckpointConfig();
-//        config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+        env.enableCheckpointing(1000);
+        env.setStateBackend(new FsStateBackend(checkpointPath));
+        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 3000));
+        CheckpointConfig config = env.getCheckpointConfig();
+        config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
         // env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
 

src/main/java/com/gmei/data/monitor/operator/PortraitMonitorErrOperator.java

@@ -62,7 +62,6 @@ public class PortraitMonitorErrOperator implements BaseOperator{
                                         jsonObject.put("statistics_action", eventType);
                                     }
                                 }
-                            }
                                 String appAction = appObject.getString("action");
                                 if (null != appAction) {
                                     String[] edits = {"create", "update", "answer"};
@@ -75,6 +74,7 @@ public class PortraitMonitorErrOperator implements BaseOperator{
                                     }
                                 }
                             }
+                        }
                         String type = jsonObject.getString("type");
                         String device = jsonObject.getString("device");
                         if (null != type && null != device) {
@@ -92,6 +92,15 @@ public class PortraitMonitorErrOperator implements BaseOperator{
                     }
                 }
             })
+            .filter(new FilterFunction<String>() {
+                  @Override
+                  public boolean filter(String value) throws Exception {
+                      if(StringUtils.isBlank(value)){
+                          return false;
+                      }
+                      return true;
+                  }
+              })
             .map(new MapFunction<String,Tuple2<String,String>>() {
                 @Override
                 public Tuple2<String,String> map(String value) throws Exception {
@@ -99,15 +108,6 @@ public class PortraitMonitorErrOperator implements BaseOperator{
                     return new Tuple2<>(value,monitorTime);
                 }
             })
-            .filter(new FilterFunction<Tuple2<String,String>>() {
-                @Override
-                public boolean filter( Tuple2<String,String> tuple2) throws Exception {
-                    if(StringUtils.isBlank(tuple2.f0)){
-                        return false;
-                    }
-                    return true;
-                }
-            })
             .addSink(new PortraitErrMysqlSink(outJdbcUrl,maxRetry,retryInteral))
             .setParallelism(parallelism);
     }