修改计算速率方式，使用session总数除以时间窗口

1 files changed, 3 insertions, 1 deletions

diff --git a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
index 464b5f6..aab3e88 100644
--- a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
@@ -80,7 +80,9 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
                 }
             }
             String sourceIpList = StringUtils.join(sourceIpSet, ",");
-            return Tuple6.of(sessions/cnt/duration,packets/cnt/duration,bytes/cnt/duration,sourceIpList,startTime,duration);
+//            return Tuple6.of(sessions/cnt/duration,packets/cnt/duration,bytes/cnt/duration,sourceIpList,startTime,duration);
+            return Tuple6.of(sessions/CommonConfig.FLINK_WINDOW_MAX_TIME,packets/CommonConfig.FLINK_WINDOW_MAX_TIME,
+                    bytes*8/CommonConfig.FLINK_WINDOW_MAX_TIME,sourceIpList,startTime,duration);
         }catch (Exception e){
             logger.error("聚合中间结果集失败 {}",e);
         }