metrics统计增加根据server IP hashcode分区数。

4 files changed, 22 insertions, 0 deletions

diff --git a/src/main/java/com/zdjizhi/common/CommonConfig.java b/src/main/java/com/zdjizhi/common/CommonConfig.java
index 63a1977..1c6824c 100644
--- a/src/main/java/com/zdjizhi/common/CommonConfig.java
+++ b/src/main/java/com/zdjizhi/common/CommonConfig.java
@@ -36,6 +36,7 @@ public class CommonConfig {
     public static final int FLINK_WINDOW_MAX_TIME = CommonConfigurations.getIntProperty("flink.window.max.time");
 
     public static final int SOURCE_IP_LIST_LIMIT = CommonConfigurations.getIntProperty("source.ip.list.limit");
+    public static final int DESTINATION_IP_PARTITION_NUM = CommonConfigurations.getIntProperty("destination.ip.partition.num");
     public static final int DATA_CENTER_ID_NUM = CommonConfigurations.getIntProperty("data.center.id.num");
 
     public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");
diff --git a/src/main/java/com/zdjizhi/common/DosMetricsLog.java b/src/main/java/com/zdjizhi/common/DosMetricsLog.java
index 13f4d96..ac5050b 100644
--- a/src/main/java/com/zdjizhi/common/DosMetricsLog.java
+++ b/src/main/java/com/zdjizhi/common/DosMetricsLog.java
@@ -12,6 +12,15 @@ public class DosMetricsLog implements Serializable {
     private long session_rate;
     private long packet_rate;
     private long bit_rate;
+    private int partition_num;
+
+    public int getPartition_num() {
+        return partition_num;
+    }
+
+    public void setPartition_num(int partition_num) {
+        this.partition_num = partition_num;
+    }
 
     public long getSketch_start_time() {
         return sketch_start_time;
@@ -88,6 +97,7 @@ public class DosMetricsLog implements Serializable {
                 ", session_rate=" + session_rate +
                 ", packet_rate=" + packet_rate +
                 ", bit_rate=" + bit_rate +
+                ", partition_num=" + partition_num +
                 '}';
     }
 }
diff --git a/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java b/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
index bb6f77f..13fd24a 100644
--- a/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
+++ b/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
@@ -21,6 +21,7 @@ class TrafficServerIpMetrics {
         dosMetricsLog.setSession_rate(midResuleLog.getSketch_sessions());
         dosMetricsLog.setPacket_rate(midResuleLog.getSketch_packets());
         dosMetricsLog.setBit_rate(midResuleLog.getSketch_bytes());
+        dosMetricsLog.setPartition_num(getPartitionNumByIp(midResuleLog.getDestination_ip()));
         logger.debug("metric 结果已加载：{}",dosMetricsLog.toString());
         return dosMetricsLog;
     }
@@ -29,4 +30,13 @@ class TrafficServerIpMetrics {
         return sketchStartTime / CommonConfig.FLINK_WINDOW_MAX_TIME * CommonConfig.FLINK_WINDOW_MAX_TIME;
     }
 
+    private static int getPartitionNumByIp(String destinationIp){
+        return Math.abs(destinationIp.hashCode()) % CommonConfig.DESTINATION_IP_PARTITION_NUM;
+    }
+
+    public static void main(String[] args) {
+//        System.out.println(getPartitionNumByIp("146.177.223.43"));
+        System.out.println("146.177.223.43".hashCode());
+    }
+
 }
diff --git a/src/main/resources/common.properties b/src/main/resources/common.properties
index 5e3d0b2..03e2f55 100644
--- a/src/main/resources/common.properties
+++ b/src/main/resources/common.properties
@@ -28,6 +28,7 @@ flink.watermark.max.orderness=1
 flink.window.max.time=10
 
 source.ip.list.limit=10000
+destination.ip.partition.num=10000
 
 data.center.id.num=15