代码格式化，实体类重写equals、hashcode方法。

src/main/java/com/zdjizhi/common/CommonConfig.java

@@ -3,7 +3,8 @@ package com.zdjizhi.common;
 import com.zdjizhi.utils.CommonConfigurations;
 
 /**
- * Created by wk on 2021/1/6.
+ * @author wlh
+ * @date 2021/1/6
  */
 public class CommonConfig {
 

src/main/java/com/zdjizhi/common/DosBaselineThreshold.java

@@ -2,6 +2,7 @@ package com.zdjizhi.common;
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Objects;
 
 public class DosBaselineThreshold implements Serializable {
     private ArrayList<Integer> session_rate;
@@ -40,4 +41,23 @@ public class DosBaselineThreshold implements Serializable {
                 ", session_rate_default_value=" + session_rate_default_value +
                 '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosBaselineThreshold)) {
+            return false;
+        }
+        DosBaselineThreshold that = (DosBaselineThreshold) o;
+        return Objects.equals(getSession_rate(), that.getSession_rate()) &&
+                Objects.equals(getSession_rate_baseline_type(), that.getSession_rate_baseline_type()) &&
+                Objects.equals(getSession_rate_default_value(), that.getSession_rate_default_value());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getSession_rate(), getSession_rate_baseline_type(), getSession_rate_default_value());
+    }
 }

src/main/java/com/zdjizhi/common/DosDetectionThreshold.java

@@ -2,7 +2,11 @@ package com.zdjizhi.common;
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Objects;
 
+/**
+ * @author wlh
+ */
 public class DosDetectionThreshold implements Serializable {
     private String profileId;
     private String attackType;
@@ -27,6 +31,30 @@ public class DosDetectionThreshold implements Serializable {
                 '}';
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        DosDetectionThreshold threshold = (DosDetectionThreshold) o;
+        return packetsPerSec == threshold.packetsPerSec &&
+                bitsPerSec == threshold.bitsPerSec &&
+                sessionsPerSec == threshold.sessionsPerSec &&
+                isValid == threshold.isValid &&
+                Objects.equals(profileId, threshold.profileId) &&
+                Objects.equals(attackType, threshold.attackType) &&
+                Objects.equals(serverIpList, threshold.serverIpList) &&
+                Objects.equals(serverIpAddr, threshold.serverIpAddr);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(profileId, attackType, serverIpList, serverIpAddr, packetsPerSec, bitsPerSec, sessionsPerSec, isValid);
+    }
+
     public String getProfileId() {
         return profileId;
     }

src/main/java/com/zdjizhi/common/DosEventLog.java

@@ -1,6 +1,7 @@
 package com.zdjizhi.common;
 
 import java.io.Serializable;
+import java.util.Objects;
 
 public class DosEventLog implements Serializable {
 
@@ -140,4 +141,33 @@ public class DosEventLog implements Serializable {
                 ", bit_rate=" + bit_rate +
                 '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosEventLog)) {
+            return false;
+        }
+        DosEventLog that = (DosEventLog) o;
+        return getLog_id() == that.getLog_id() &&
+                getStart_time() == that.getStart_time() &&
+                getEnd_time() == that.getEnd_time() &&
+                getSession_rate() == that.getSession_rate() &&
+                getPacket_rate() == that.getPacket_rate() &&
+                getBit_rate() == that.getBit_rate() &&
+                Objects.equals(getAttack_type(), that.getAttack_type()) &&
+                Objects.equals(getSeverity(), that.getSeverity()) &&
+                Objects.equals(getConditions(), that.getConditions()) &&
+                Objects.equals(getDestination_ip(), that.getDestination_ip()) &&
+                Objects.equals(getDestination_country(), that.getDestination_country()) &&
+                Objects.equals(getSource_ip_list(), that.getSource_ip_list()) &&
+                Objects.equals(getSource_country_list(), that.getSource_country_list());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getLog_id(), getStart_time(), getEnd_time(), getAttack_type(), getSeverity(), getConditions(), getDestination_ip(), getDestination_country(), getSource_ip_list(), getSource_country_list(), getSession_rate(), getPacket_rate(), getBit_rate());
+    }
 }

src/main/java/com/zdjizhi/common/DosMetricsLog.java

@@ -1,6 +1,7 @@
 package com.zdjizhi.common;
 
 import java.io.Serializable;
+import java.util.Objects;
 
 public class DosMetricsLog implements Serializable {
 
@@ -80,4 +81,27 @@ public class DosMetricsLog implements Serializable {
                 ", partition_num=" + partition_num +
                 '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosMetricsLog)) {
+            return false;
+        }
+        DosMetricsLog that = (DosMetricsLog) o;
+        return getSketch_start_time() == that.getSketch_start_time() &&
+                getSession_rate() == that.getSession_rate() &&
+                getPacket_rate() == that.getPacket_rate() &&
+                getBit_rate() == that.getBit_rate() &&
+                getPartition_num() == that.getPartition_num() &&
+                Objects.equals(getAttack_type(), that.getAttack_type()) &&
+                Objects.equals(getDestination_ip(), that.getDestination_ip());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getSketch_start_time(), getAttack_type(), getDestination_ip(), getSession_rate(), getPacket_rate(), getBit_rate(), getPartition_num());
+    }
 }

src/main/java/com/zdjizhi/common/DosSketchLog.java

@@ -1,6 +1,7 @@
 package com.zdjizhi.common;
 
 import java.io.Serializable;
+import java.util.Objects;
 
 public class DosSketchLog implements Serializable {
 
@@ -31,6 +32,32 @@ public class DosSketchLog implements Serializable {
                 '}';
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DosSketchLog)) {
+            return false;
+        }
+        DosSketchLog sketchLog = (DosSketchLog) o;
+        return getSketch_start_time() == sketchLog.getSketch_start_time() &&
+                getSketch_duration() == sketchLog.getSketch_duration() &&
+                getSketch_sessions() == sketchLog.getSketch_sessions() &&
+                getSketch_packets() == sketchLog.getSketch_packets() &&
+                getSketch_bytes() == sketchLog.getSketch_bytes() &&
+                Objects.equals(getCommon_sled_ip(), sketchLog.getCommon_sled_ip()) &&
+                Objects.equals(getCommon_data_center(), sketchLog.getCommon_data_center()) &&
+                Objects.equals(getAttack_type(), sketchLog.getAttack_type()) &&
+                Objects.equals(getSource_ip(), sketchLog.getSource_ip()) &&
+                Objects.equals(getDestination_ip(), sketchLog.getDestination_ip());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getCommon_sled_ip(), getCommon_data_center(), getSketch_start_time(), getSketch_duration(), getAttack_type(), getSource_ip(), getDestination_ip(), getSketch_sessions(), getSketch_packets(), getSketch_bytes());
+    }
+
     public String getCommon_sled_ip() {
         return common_sled_ip;
     }

src/main/java/com/zdjizhi/etl/ParseSketchLog.java

@@ -69,7 +69,7 @@ public class ParseSketchLog {
                         dosSketchLog.setSketch_packets(sketchPackets);
                         dosSketchLog.setSketch_bytes(sketchBytes);
                         collector.collect(dosSketchLog);
-                        logger.debug("数据解析成功：{}",dosSketchLog.toString());
+                        logger.info("数据解析成功：{}",dosSketchLog.toString());
                     }
                 }
             } catch (Exception e) {

src/main/resources/common.properties

@@ -11,8 +11,8 @@ kafka.input.parallelism=1
 kafka.input.topic.name=DOS-SKETCH-RECORD
 
 #输入kafka地址
-#kafka.input.bootstrap.servers=192.168.44.12:9092
+kafka.input.bootstrap.servers=192.168.44.12:9094
-kafka.input.bootstrap.servers=192.168.44.11:9094,192.168.44.14:9094,192.168.44.15:9094
+#kafka.input.bootstrap.servers=192.168.44.11:9094,192.168.44.14:9094,192.168.44.15:9094
 
 #读取kafka group id
 kafka.input.group.id=2109160928
@@ -60,7 +60,7 @@ flink.detection.map.parallelism=1
 flink.watermark.max.orderness=10
 
 #计算窗口大小，默认600s
-flink.window.max.time=600
+flink.window.max.time=10
 
 #dos event结果中distinct source IP限制
 source.ip.list.limit=10000