update codes

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

@@ -58,17 +58,17 @@ public class PortraitMonitorMain {
         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,
@@ -92,7 +92,7 @@ public class PortraitMonitorMain {
         ).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();
 

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

@@ -69,16 +69,11 @@ public class PortraitMonitorShdOperator implements BaseOperator{
                 if (StringUtils.isNotBlank(maidianEventTime)) {
                     logTime = Long.valueOf(maidianEventTime) * 1000;
                 }
-                String backendEventTime = jsonObject.getString("TIME");
-                if (StringUtils.isNotBlank(backendEventTime)) {
-                    try {
-                        logTime = dateTimeFormat.parseMillis(backendEventTime);
-                    } catch (IllegalArgumentException e) {
-                        try {
-                            logTime = dateTimeNoMillisFormat.parseMillis(backendEventTime);
-                        } catch (IllegalArgumentException e2) {
-                            e2.printStackTrace();
-                        }
+                JSONObject appObject = jsonObject.getJSONObject("APP");
+                if (null != appObject) {
+                    Long time  = appObject.getLong("time");
+                    if(null != time && time != 0){
+                        logTime = time;
                     }
                 }
                 return logTime;