update codes

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

@@ -36,7 +36,7 @@ public class PortraitMonitorMain {
         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",3);
+        Integer windowSize = parameterTool.getInt("windowSize",60000);
         Integer slideSize = parameterTool.getInt("slideSize",3);
         String outJdbcUrl = parameterTool.get("outJdbcUrl",
                 "jdbc:mysql://test002:3306/flink_monitor?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
@@ -49,13 +49,14 @@ public class PortraitMonitorMain {
 
         // 获得流处理环境对象
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
         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);
+
 
         // 获取数据源
         DataStream portraitErrDataStream = new PortraitKafkaSource(

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

@@ -8,8 +8,11 @@ import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.datastream.DataStream;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.ISODateTimeFormat;
 
 import java.util.Arrays;
+import java.util.Date;
 
 /**
  * @ClassName PortraitMonitorErr
@@ -24,6 +27,8 @@ public class PortraitMonitorErrOperator implements BaseOperator{
     private int maxRetry;
     private long retryInteral;
     private int parallelism;
+    public static final DateTimeFormatter dateTimeFormat = ISODateTimeFormat.dateTime();
+    public static final DateTimeFormatter dateTimeNoMillisFormat = ISODateTimeFormat.dateTimeNoMillis();
 
     public PortraitMonitorErrOperator(DataStream dataStream, String outJdbcUrl, int maxRetry, long retryInteral,int parallelism) {
         this.dataStream = dataStream;
@@ -104,8 +109,26 @@ public class PortraitMonitorErrOperator implements BaseOperator{
             .map(new MapFunction<String,Tuple2<String,String>>() {
                 @Override
                 public Tuple2<String,String> map(String value) throws Exception {
-                    String monitorTime = DateUtils.getCurrentTimeStr();
-                    return new Tuple2<>(value,monitorTime);
+                    long logTime = 0;
+                    JSONObject jsonObject = JSONObject.parseObject(value);
+                    String maidianEventTime = jsonObject.getString("create_at");
+                    if(StringUtils.isNotBlank(maidianEventTime)){
+                        logTime = Long.valueOf(maidianEventTime);
+                    }
+                    String backendEventTime = jsonObject.getString("TIME");
+                    if(StringUtils.isNotBlank(backendEventTime)){
+                        try {
+                            logTime = dateTimeFormat.parseMillis(backendEventTime)/1000;
+                        } catch (IllegalArgumentException e) {
+                            try {
+                                logTime = dateTimeNoMillisFormat.parseMillis(backendEventTime)/1000;
+                            } catch (IllegalArgumentException e2) {
+                                e2.printStackTrace();
+                            }
+                        }
+                    }
+                    String eventTime = DateUtils.secondToDate(logTime);
+                    return new Tuple2<>(value,eventTime);
                 }
             })
             .addSink(new PortraitErrMysqlSink(outJdbcUrl,maxRetry,retryInteral))

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

 package com.gmei.data.monitor.operator;
 
 import com.alibaba.fastjson.JSONObject;
+import com.gmei.data.monitor.bean.GmPortraitResult;
 import com.gmei.data.monitor.bean.TblMonitorPortraitShd;
 import com.gmei.data.monitor.sink.PortraitShdMysqlSink;
 import com.gmei.data.monitor.utils.DateUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.KeyedStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.datastream.WindowedStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.ISODateTimeFormat;
 
 import java.util.Arrays;
+import java.util.Date;
 
 /**
  * @ClassName PortraitMonitorShdOperator
@@ -33,6 +40,8 @@ public class PortraitMonitorShdOperator implements BaseOperator{
     private int maxRetry;
     private long retryInteral;
     private int parallelism;
+    public static final DateTimeFormatter dateTimeFormat = ISODateTimeFormat.dateTime();
+    public static final DateTimeFormatter dateTimeNoMillisFormat = ISODateTimeFormat.dateTimeNoMillis();
 
     public PortraitMonitorShdOperator(DataStream dataStream, int windownSize,int slideSize,String outJdbcUrl, int maxRetry, long retryInteral,int parallelism) {
         this.dataStream = dataStream;
@@ -46,20 +55,20 @@ public class PortraitMonitorShdOperator implements BaseOperator{
 
     @Override
     public void run() {
-        dataStream
-            .map(new MapFunction<String, JSONObject>() {
+        SingleOutputStreamOperator map01 = dataStream.map(new MapFunction<String, JSONObject>() {
             @Override
             public JSONObject map(String value) throws Exception {
                 JSONObject jsonObject = new JSONObject();
-                    try{
+                try {
                     jsonObject = JSONObject.parseObject(value);
-                    }catch (Exception e){
+                } catch (Exception e) {
                     e.printStackTrace();
                 }
                 return jsonObject;
             }
-            })
-            .filter(new FilterFunction<JSONObject>() {
+        });
+        //map01.print();
+        SingleOutputStreamOperator filter01 = map01.filter(new FilterFunction<JSONObject>() {
             @Override
             public boolean filter(JSONObject jsonObject) throws Exception {
                 try {
@@ -75,7 +84,8 @@ public class PortraitMonitorShdOperator implements BaseOperator{
                                 return true;
                             }
                         }
-                            JSONObject appObject = sysObject.getJSONObject("APP");
+                    }
+                    JSONObject appObject = jsonObject.getJSONObject("APP");
                     if (null != appObject) {
                         String eventType = appObject.getString("event_type");
                         if (null != eventType) {
@@ -100,7 +110,6 @@ public class PortraitMonitorShdOperator implements BaseOperator{
                             }
                         }
                     }
-                        }
                     String type = jsonObject.getString("type");
                     String device = jsonObject.getString("device");
                     if (null != type && null != device) {
@@ -118,30 +127,58 @@ public class PortraitMonitorShdOperator implements BaseOperator{
                 }
                 return false;
             }
-            })
-            .map(new MapFunction<JSONObject, Tuple2<String, JSONObject>>() {
+        });
+        //filter01.print();
+        SingleOutputStreamOperator map02 = filter01.map(new MapFunction<JSONObject, Tuple2<String, JSONObject>>() {
             @Override
             public Tuple2<String, JSONObject> map(JSONObject jsonObject) throws Exception {
                 String statisticsAction = jsonObject.getString("statistics_action");
                 return new Tuple2<String, JSONObject>(statisticsAction, jsonObject) {
                 };
             }
-            })
-            .keyBy(0)
-            .timeWindow(Time.seconds(windownSize),Time.seconds(slideSize))
-            .process(new ProcessWindowFunction<Tuple2<String,JSONObject>, TblMonitorPortraitShd, Tuple, TimeWindow>() {
+        });
+        //map02.print();
+        KeyedStream keyedStream = map02.keyBy(0);
+        //keyedStream.print();
+        WindowedStream windowedStream = keyedStream.timeWindow(Time.seconds(windownSize), Time.seconds(slideSize));
+        SingleOutputStreamOperator process = windowedStream.process(new ProcessWindowFunction<Tuple2<String, JSONObject>, TblMonitorPortraitShd, Tuple, TimeWindow>() {
             @Override
-                public void process(Tuple key, Context context, Iterable<Tuple2<String,JSONObject>> elements, Collector<TblMonitorPortraitShd> out) {
+            public void process(Tuple key, Context context, Iterable<Tuple2<String, JSONObject>> elements, Collector<TblMonitorPortraitShd> out) {
                 Integer count = 0;
-                    for (Tuple2<String,JSONObject> tuple2 : elements) {
-                        ++ count;
+                Date date = new Date();
+                long currentTimestamp = DateUtils.getCurrentTimestamp(date);
+                long tenMinitesAgoTimestamp = DateUtils.getTenMinitesAgoTimestamp(date);
+                for (Tuple2<String, JSONObject> tuple2 : elements) {
+                    System.err.println(tuple2.f0);
+                    long logTime = 0;
+                    JSONObject jsonObject = tuple2.f1;
+                    String maidianEventTime = jsonObject.getString("create_at");
+                    if (StringUtils.isNotBlank(maidianEventTime)) {
+                        logTime = Long.valueOf(maidianEventTime);
+                    }
+                    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();
+                            }
+                        }
+                    }
+                    if (logTime >= tenMinitesAgoTimestamp && logTime <= currentTimestamp) {
+                        ++count;
+                    }
                 }
-                    String monitorTime = DateUtils.getCurrentTimeStr();
-                    TblMonitorPortraitShd tblMonitorPortraitShd = new TblMonitorPortraitShd(key.getField(0), count, monitorTime);
+                String currentTimeStr = DateUtils.getCurrentTimeStr(date);
+                TblMonitorPortraitShd tblMonitorPortraitShd = new TblMonitorPortraitShd(key.getField(0), count, currentTimeStr);
                 out.collect(tblMonitorPortraitShd);
             }
-            })
-            .addSink(new PortraitShdMysqlSink(outJdbcUrl,maxRetry,retryInteral))
+        });
+        process.print();
+        process.addSink(new PortraitShdMysqlSink(outJdbcUrl,maxRetry,retryInteral))
                 .setParallelism(parallelism);
     }
 }

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

@@ -16,6 +16,8 @@ import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 
+import java.util.Date;
+
 /**
  * @ClassName PortraitMonitorSucOperator
  * @Description: 用户画像成功打点数监控
@@ -77,11 +79,17 @@ public class PortraitMonitorSucOperator implements BaseOperator{
                 @Override
                 public void process(Tuple key, Context context, Iterable<GmPortraitResult> elements, Collector<TblMonitorPortraitSuc> out) {
                     Integer count = 0;
+                    Date date = new Date();
+                    long currentTimestamp = DateUtils.getCurrentTimestamp(date);
+                    long tenMinitesAgoTimestamp = DateUtils.getTenMinitesAgoTimestamp(date);
                     for (GmPortraitResult element : elements) {
+                        long logTime = Long.valueOf(element.getLog_time());
+                        if(logTime >= tenMinitesAgoTimestamp && logTime <= currentTimestamp){
                             ++ count;
                         }
-                    String monitorTime = DateUtils.getCurrentTimeStr();
-                    TblMonitorPortraitSuc tblMonitorPortraitSuc = new TblMonitorPortraitSuc(key.getField(0), count, monitorTime);
+                    }
+                    String currentTimeStr = DateUtils.getCurrentTimeStr(date);
+                    TblMonitorPortraitSuc tblMonitorPortraitSuc = new TblMonitorPortraitSuc(key.getField(0), count, currentTimeStr);
                     out.collect(tblMonitorPortraitSuc);
                 }
             })

src/main/java/com/gmei/data/monitor/utils/DateUtils.java

@@ -2,6 +2,7 @@ package com.gmei.data.monitor.utils;
 
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
+import java.util.Calendar;
 import java.util.Date;
 
 /**
@@ -17,10 +18,45 @@ public class DateUtils {
 
     /**
      * 获取当前时间字符串
+     * @param date
      * @return
      */
-    public static String getCurrentTimeStr() {
-        return new SimpleDateFormat(DATE_FORMATE_YMDHMS).format(new Date());
+    public static String getCurrentTimeStr(Date date) {
+        return new SimpleDateFormat(DATE_FORMATE_YMDHMS).format(date);
+    }
+
+    /**
+     * 获取十分钟以前的时间字符串
+     * @param date
+     * @return
+     */
+    public static String getTenMinitesAgoTimeStr(Date date) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTime(date);
+        calendar.add(Calendar.MINUTE,-10);
+        SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMATE_YMDHMS);
+        return sdf.format(calendar.getTime());
+    }
+
+    /**
+     * 获取当前时间戳
+     * @param date
+     * @return
+     */
+    public static long getCurrentTimestamp(Date date) {
+        return date.getTime();
+    }
+
+    /**
+     * 获取十分钟以前的时间戳
+     * @param date
+     * @return
+     */
+    public static long getTenMinitesAgoTimestamp(Date date) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTime(date);
+        calendar.add(Calendar.MINUTE,-10);
+        return calendar.getTime().getTime();
     }
 
     /**
@@ -33,4 +69,18 @@ public class DateUtils {
         SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_FORMATE_YMDHMS);
         return simpleDateFormat.parse(dateStr).getTime();
     }
+
+    /**
+     * 将秒值转为指定格式的日期
+     * @param second
+     * @return
+     */
+    public static String secondToDate(long second) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.setTimeInMillis(second * 1000);
+        Date date = calendar.getTime();
+        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMATE_YMDHMS);
+        String dateString = format.format(date);
+        return dateString;
+    }
 }