update codes

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

@@ -48,11 +48,17 @@ public class PortraitMonitorSucOperator implements BaseOperator{
 
     @Override
     public void run() {
-        SingleOutputStreamOperator map = dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
+        SingleOutputStreamOperator map = dataStream.filter(new FilterFunction<String>() {
+            @Override
+            public boolean filter(String value) throws Exception {
+                return JSON.isValid(value);
+            }
+        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
             @Override
             public long extractAscendingTimestamp(String element) {
                 JSONObject jsonObject = JSON.parseObject(element);
-                Long dateTime = jsonObject.getLong("log_time");
+                String dateTimeStr = jsonObject.getString("log_time");
+                long dateTime = Math.round(Double.valueOf(dateTimeStr));
                 return dateTime;
             }
         }).map(new MapFunction<String, GmPortraitResult>() {