中心测试版本，修改为在clickhouse计算后读取结果

IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java

@@ -41,9 +41,11 @@ public class BaseArangoData {
                 map.put(i,new ConcurrentHashMap<>());
             }
             CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
-            long[] timeRange = getTimeRange(table);
+//            long[] timeRange = getTimeRange(table);
+            Long total = getCountTotal(table);
             for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
-                String sql = getQuerySql(timeRange, i, table);
+//                String sql = getQuerySql(timeRange, i, table);
+                String sql = getQuerySql(total, i, table);
                 ReadHistoryArangoData<T> readHistoryArangoData =
                         new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
                 threadPool.executor(readHistoryArangoData);
@@ -56,6 +58,30 @@ public class BaseArangoData {
         }
     }
 
+    private Long getCountTotal(String table){
+        long start = System.currentTimeMillis();
+        Long cnt = 0L;
+        String sql = "RETURN LENGTH("+table+")";
+        try {
+            ArangoCursor<Long> longs = arangoDBConnect.executorQuery(sql, Long.class);
+            while (longs.hasNext()){
+                cnt = longs.next();
+            }
+        }catch (Exception e){
+            LOG.error(sql +"执行异常");
+        }
+        long last = System.currentTimeMillis();
+        LOG.info(sql+" 结果："+cnt+" 执行时间："+(last-start));
+        return cnt;
+    }
+
+    private String getQuerySql(Long cnt,int threadNumber, String table){
+        long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER + 1;
+        long offsetNum = threadNumber * sepNum;
+        return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
+    }
+
+
     private long[] getTimeRange(String table){
         long minTime = 0L;
         long maxTime = 0L;

IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java

@@ -44,14 +44,14 @@ public class UpdateGraphData {
         long start = System.currentTimeMillis();
         try {
 
-            updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN", Fqdn.class,BaseDocument.class,
-                    ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
+//            updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN", Fqdn.class,BaseDocument.class,
+//                    ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
 
-            updateDocument(newVertexIpMap,historyVertexIpMap,"IP", Ip.class,BaseDocument.class,
-                    ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);
+//            updateDocument(newVertexIpMap,historyVertexIpMap,"IP", Ip.class,BaseDocument.class,
+//                    ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);
 
-            updateDocument(newVertexSubscriberMap,historyVertexSubscriberMap,"SUBSCRIBER", Subscriber.class,BaseDocument.class,
-                    ReadClickhouseData::getVertexSubscriberSql,ReadClickhouseData::getVertexSubscriberDocument);
+//            updateDocument(newVertexSubscriberMap,historyVertexSubscriberMap,"SUBSCRIBER", Subscriber.class,BaseDocument.class,
+//                    ReadClickhouseData::getVertexSubscriberSql,ReadClickhouseData::getVertexSubscriberDocument);
 
             updateDocument(newRelationFqdnAddressIpMap,historyRelationFqdnAddressIpMap,"R_LOCATE_FQDN2IP", LocateFqdn2Ip.class,BaseEdgeDocument.class,
                     ReadClickhouseData::getRelationshipFqdnAddressIpSql,ReadClickhouseData::getRelationFqdnAddressIpDocument);
@@ -60,9 +60,9 @@ public class UpdateGraphData {
 //                    VisitIp2Fqdn.class,BaseEdgeDocument.class,
 //                    ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument);
 
-            updateDocument(newRelationSubsciberLocateIpMap,historyRelationSubsciberLocateIpMap,"R_LOCATE_SUBSCRIBER2IP",
-                    LocateSubscriber2Ip.class,BaseEdgeDocument.class,
-                    ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument);
+//            updateDocument(newRelationSubsciberLocateIpMap,historyRelationSubsciberLocateIpMap,"R_LOCATE_SUBSCRIBER2IP",
+//                    LocateSubscriber2Ip.class,BaseEdgeDocument.class,
+//                    ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument);
 
 
             long last = System.currentTimeMillis();

IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java

@@ -270,9 +270,9 @@ public class ReadClickhouseData {
     }
 
     public static String getRelationshipFqdnAddressIpSql() {
-        String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
-        String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'TLS' AS  schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip";
-        String httpSql = "SELECT http_host AS FQDN,common_server_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip";
+        String where = " common_end_time >= " + minTime + " AND common_end_time < " + maxTime;
+        String sslSql = "SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'TLS' AS  schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip";
+        String httpSql = "SELECT http_host AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(common_client_ip) AS DIST_CIP_RECENT,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip";
         return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
     }
 

IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadHistoryArangoData.java

@@ -97,6 +97,11 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
     private void deleteDistinctClientIpByTime(T doc) {
         ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
         ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
+        if (distCip == null || distCip.isEmpty()){
+            doc.updateAttribute("DIST_CIP", new String[0]);
+            doc.updateAttribute("DIST_CIP_TS", new long[0]);
+            return;
+        }
         distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
         Collections.sort(distCipTs);
         Collections.reverse(distCipTs);

IP-learning-graph/src/main/resources/application.properties

@@ -4,7 +4,7 @@ arangoDB.host=192.168.40.182
 arangoDB.port=8529
 arangoDB.user=upsert
 arangoDB.password=ceiec2018
-arangoDB.DB.name=ip-learning-test
+arangoDB.DB.name=ip-learning-test-0
 #arangoDB.DB.name=tsg_galaxy_v3
 arangoDB.batch=100000
 arangoDB.ttl=3600
@@ -12,16 +12,16 @@ arangoDB.ttl=3600
 arangoDB.read.limit=
 update.arango.batch=10000
 
-thread.pool.number=10
+thread.pool.number=80
 thread.await.termination.time=10
 
 
 #读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
-time.limit.type=0
-read.clickhouse.max.time=1596684142
-read.clickhouse.min.time=1596425769
+time.limit.type=1
+read.clickhouse.max.time=1598246519
+read.clickhouse.min.time=1597161600
 
 update.interval=3600
-distinct.client.ip.num=10000
+distinct.client.ip.num=1
 recent.count.hour=24
 

ip-learning-spark/src/main/java/cn/ac/iie/dao/BaseArangoData.java

@@ -33,9 +33,8 @@ public class BaseArangoData {
 
     private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
 
-    public <T extends BaseDocument> void readHistoryData(String table,
-                                                  ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
-                                                  Class<T> type) {
+    public <T extends BaseDocument> ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> readHistoryData(String table, Class<T> type) {
+        ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap = new ConcurrentHashMap<>();
         try {
             LOG.warn("开始更新" + table);
             long start = System.currentTimeMillis();
@@ -43,9 +42,11 @@ public class BaseArangoData {
                 historyMap.put(i, new ConcurrentHashMap<>());
             }
             CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER());
-            long[] timeRange = getTimeRange(table);
+//            long[] timeRange = getTimeRange(table);
+            Long total = getCountTotal(table);
             for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
-                String sql = getQuerySql(timeRange, i, table);
+//                String sql = getQuerySql(timeRange, i, table);
+                String sql = getQuerySql(total, i, table);
                 ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
                 threadPool.executor(readHistoryArangoData);
             }
@@ -55,6 +56,30 @@ public class BaseArangoData {
         } catch (Exception e) {
             e.printStackTrace();
         }
+        return historyMap;
+    }
+
+    private Long getCountTotal(String table){
+        long start = System.currentTimeMillis();
+        Long cnt = 0L;
+        String sql = "RETURN LENGTH("+table+")";
+        try {
+            ArangoCursor<Long> longs = arangoDBConnect.executorQuery(sql, Long.class);
+            while (longs.hasNext()){
+                cnt = longs.next();
+            }
+        }catch (Exception e){
+            LOG.error(sql +"执行异常");
+        }
+        long last = System.currentTimeMillis();
+        LOG.warn(sql+" 结果："+cnt+" 执行时间："+(last-start));
+        return cnt;
+    }
+
+    private String getQuerySql(Long cnt,int threadNumber, String table){
+        long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER() + 1;
+        long offsetNum = threadNumber * sepNum;
+        return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
     }
 
     private long[] getTimeRange(String table) {

ip-learning-spark/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java

@@ -58,6 +58,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
     public void run() {
         try {
             long s = System.currentTimeMillis();
+            LOG.warn(query+" \n 开始查询");
             ArangoCursor<T> docs = arangoConnect.executorQuery(query, type);
             if (docs != null) {
                 List<T> baseDocuments = docs.asListRemaining();
@@ -69,9 +70,9 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                             updateProtocolDocument(doc);
                             deleteDistinctClientIpByTime(doc);
                             break;
-                        case "R_VISIT_IP2FQDN":
-                            updateProtocolDocument(doc);
-                            break;
+//                        case "R_VISIT_IP2FQDN":
+//                            updateProtocolDocument(doc);
+//                            break;
                         default:
                     }
                     int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER();
@@ -107,6 +108,12 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
     private void deleteDistinctClientIpByTime(T doc) {
         ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
         ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
+        if (distCip == null || distCip.isEmpty()){
+            doc.updateAttribute("DIST_CIP", new String[0]);
+            doc.updateAttribute("DIST_CIP_TS", new long[0]);
+            return;
+        }
+
         distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
         Collections.sort(distCipTs);
         int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);

ip-learning-spark/src/main/resources/application.properties

@@ -7,13 +7,16 @@ repartitionNumber=36
 spark.serializer=org.apache.spark.serializer.KryoSerializer
 master=local[*]
 #spark读取clickhouse配置
-spark.read.clickhouse.url=jdbc:clickhouse://192.168.40.186:8123/tsg_galaxy_v3
+#spark.read.clickhouse.url=jdbc:clickhouse://192.168.40.186:8123/tsg_galaxy_v3
+spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.12:8123/tsg_galaxy_v3
 spark.read.clickhouse.driver=ru.yandex.clickhouse.ClickHouseDriver
 spark.read.clickhouse.user=default
-spark.read.clickhouse.password=111111
+#spark.read.clickhouse.password=111111
+spark.read.clickhouse.password=ceiec2019
 spark.read.clickhouse.numPartitions=144
 spark.read.clickhouse.fetchsize=10000
-spark.read.clickhouse.partitionColumn=common_recv_time
+#spark.read.clickhouse.partitionColumn=common_end_time
+spark.read.clickhouse.partitionColumn=FIRST_FOUND_TIME
 clickhouse.socket.timeout=300000
 #arangoDB配置
 arangoDB.host=192.168.40.182
@@ -27,19 +30,19 @@ arangoDB.ttl=3600
 thread.pool.number=5
 
 #读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
-clickhouse.time.limit.type=0
-read.clickhouse.max.time=1571245220
-read.clickhouse.min.time=1571245210
+clickhouse.time.limit.type=1
+read.clickhouse.max.time=1598246519
+read.clickhouse.min.time=1597161600
 
 #读取arangoDB时间范围方式，0：正常读；1：指定时间范围
 arango.time.limit.type=0
-read.arango.max.time=1571245320
-read.arango.min.time=1571245200
+read.arango.max.time=1598246519
+read.arango.min.time=1597161600
 
 arangoDB.read.limit=
 update.arango.batch=10000
 
-distinct.client.ip.num=10000
+distinct.client.ip.num=1
 recent.count.hour=24
 
-update.interval=10800
+update.interval=3600

ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala

@@ -11,7 +11,7 @@ object BaseClickhouseData {
   val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60
   private val timeLimit: (Long, Long) = getTimeLimit
 
-  private def initClickhouseData(sql:String): Unit ={
+  private def initClickhouseData(sql:String): DataFrame ={
 
     val dataFrame: DataFrame = spark.read.format("jdbc")
       .option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL)
@@ -28,14 +28,15 @@ object BaseClickhouseData {
       .load()
     dataFrame.printSchema()
     dataFrame.createOrReplaceGlobalTempView("dbtable")
+    dataFrame
   }
 
   def loadConnectionDataFromCk(): Unit ={
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+    val where = "common_end_time >= " + timeLimit._2 + " AND common_end_time < " + timeLimit._1 + " and common_schema_type != 'BASE'"
     val sql =
       s"""
         |(SELECT
-        | ssl_sni,http_host,common_client_ip,common_server_ip,common_recv_time,common_c2s_byte_num,common_s2c_byte_num,common_schema_type
+        | ssl_sni,http_host,common_client_ip,common_server_ip,common_end_time,common_c2s_byte_num,common_s2c_byte_num,common_schema_type
         |FROM
         | connection_record_log
         |WHERE $where) as dbtable
@@ -45,6 +46,28 @@ object BaseClickhouseData {
     initClickhouseData(sql)
   }
 
+  def getRelationFqdnLocateIpDf(): DataFrame ={
+    val where = "common_end_time >= " + timeLimit._2 + " AND common_end_time < " + timeLimit._1 + " and common_schema_type != 'BASE'"
+    val sql =
+      s"""
+         |(SELECT * FROM
+         |((SELECT ssl_sni AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
+         |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'TLS' AS  schema_type
+         |FROM tsg_galaxy_v3.connection_record_log
+         |WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip)
+         |UNION ALL
+         |(SELECT http_host AS FQDN,common_server_ip,MAX(common_end_time) AS LAST_FOUND_TIME,MIN(common_end_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
+         |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'HTTP' AS schema_type
+         |FROM tsg_galaxy_v3.connection_record_log
+         |WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip))
+         |WHERE FQDN != '') as dbtable
+      """.stripMargin
+    LOG.warn(sql)
+    val frame = initClickhouseData(sql)
+    frame.printSchema()
+    frame
+  }
+
   private def loadRadiusDataFromCk(): Unit ={
     val where =
       s"""
@@ -146,6 +169,7 @@ object BaseClickhouseData {
     vertexIpDf
   }
 
+  /*
   def getRelationFqdnLocateIpDf: DataFrame ={
     loadConnectionDataFromCk()
     val sslSql =
@@ -153,10 +177,9 @@ object BaseClickhouseData {
         |SELECT
         | ssl_sni AS FQDN,
         | common_server_ip,
-        | MAX(common_recv_time) AS LAST_FOUND_TIME,
-        | MIN(common_recv_time) AS FIRST_FOUND_TIME,
+        | MAX(common_end_time) AS LAST_FOUND_TIME,
+        | MIN(common_end_time) AS FIRST_FOUND_TIME,
         | COUNT(*) AS COUNT_TOTAL,
-        | collect_set(common_client_ip) AS DIST_CIP_RECENT,
         | 'TLS' AS  schema_type
         |FROM
         | global_temp.dbtable
@@ -171,10 +194,9 @@ object BaseClickhouseData {
         |SELECT
         | http_host AS FQDN,
         | common_server_ip,
-        | MAX(common_recv_time) AS LAST_FOUND_TIME,
-        | MIN(common_recv_time) AS FIRST_FOUND_TIME,
+        | MAX(common_end_time) AS LAST_FOUND_TIME,
+        | MIN(common_end_time) AS FIRST_FOUND_TIME,
         | COUNT(*) AS COUNT_TOTAL,
-        | collect_set(common_client_ip) AS DIST_CIP_RECENT,
         | 'HTTP' AS schema_type
         |FROM
         | global_temp.dbtable
@@ -190,6 +212,7 @@ object BaseClickhouseData {
     relationFqdnLocateIpDf.printSchema()
     relationFqdnLocateIpDf
   }
+  */
 
   private def getTimeLimit: (Long,Long) ={
     var maxTime = 0L

ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala

@@ -35,7 +35,7 @@ object MergeDataFrame {
   }
 
   def mergeRelationFqdnLocateIp(): RDD[Row] ={
-    val frame = BaseClickhouseData.getRelationFqdnLocateIpDf.filter(row => isDomain(row.getAs[String]("FQDN")))
+    val frame = BaseClickhouseData.getRelationFqdnLocateIpDf().filter(row => isDomain(row.getAs[String]("FQDN")))
       .groupBy("FQDN", "common_server_ip")
       .agg(
         min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
@@ -58,13 +58,8 @@ object MergeDataFrame {
       if (fqdn == null || fqdn.length == 0) {
         return false
       }
-      if (fqdn.contains(":")) {
-        val s = fqdn.split(":")(0)
-        if (s.contains(":")){
-          return false
-        }
-      }
-      val fqdnArr = fqdn.split("\\.")
+      val domain = fqdn.split(":")(0)
+      val fqdnArr = domain.split("\\.")
       if (fqdnArr.length < 4 || fqdnArr.length > 4){
         return true
       }
@@ -80,7 +75,7 @@ object MergeDataFrame {
       }
     } catch {
       case e: Exception =>
-        LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString)
+        LOG.warn("解析域名 " + fqdn + " 失败：\n" + e.toString)
     }
     false
   }

ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocHandler.scala

@@ -93,8 +93,14 @@ object UpdateDocHandler {
     doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",",""))
   }
 
-  def mergeDistinctIp(distCipRecent:ofRef[ofRef[String]]): Array[String] ={
-     distCipRecent.flatten.distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
+  def mergeDistinctIp(distCipRecent:ofRef[String]): Array[String] ={
+    distCipRecent.flatMap(str => {
+      str.replaceAll("\\[", "")
+        .replaceAll("\\]", "")
+        .replaceAll("'", "")
+        .split(",")
+    }).distinct.toArray
+//     distCipRecent.flatten.distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
   }
 
   def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={

ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala

@@ -26,9 +26,7 @@ object UpdateDocument {
 
   def update(): Unit = {
     try {
-      updateDocument("FQDN", historyVertexFqdnMap, getVertexFqdnRow, classOf[BaseDocument], mergeVertexFqdn)
-      updateDocument("IP", historyVertexIpMap, getVertexIpRow, classOf[BaseDocument], mergeVertexIp)
-      updateDocument("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
+      updateDocument("R_LOCATE_FQDN2IP", getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
     } catch {
       case e: Exception => e.printStackTrace()
     } finally {
@@ -39,13 +37,13 @@ object UpdateDocument {
   }
 
   private def updateDocument[T <: BaseDocument](collName: String,
-                                                historyMap: ConcurrentHashMap[Integer, ConcurrentHashMap[String, T]],
                                                 getDocumentRow: (Row, ConcurrentHashMap[String, T]) => T,
                                                 clazz: Class[T],
                                                 getNewDataRdd: () => RDD[Row]
                                                ): Unit = {
-    baseArangoData.readHistoryData(collName, historyMap, clazz)
+    val historyMap = baseArangoData.readHistoryData(collName, clazz)
     val hisBc = spark.sparkContext.broadcast(historyMap)
+    LOG.warn("广播变量发送完毕")
     try {
       val start = System.currentTimeMillis()
       val newDataRdd = getNewDataRdd()
@@ -133,17 +131,18 @@ object UpdateDocument {
     val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
     val countTotalList = row.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST")
     val schemaTypeList = row.getAs[ofRef[AnyRef]]("schema_type_list")
-    val distCipRecent = row.getAs[ofRef[ofRef[String]]]("DIST_CIP_RECENT")
+
+    val distCipRecent = row.getAs[ofRef[String]]("DIST_CIP_RECENT")
+    val disCips = mergeDistinctIp(distCipRecent)
 
     val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
-    val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
 
     val key = fqdn.concat("-" + serverIp)
     var document = dictionaryMap.getOrDefault(key, null)
     if (document != null) {
       updateMaxAttribute(document, lastFoundTime, "LAST_FOUND_TIME")
       updateProtocolAttritube(document, sepAttritubeMap)
-      updateDistinctIp(document, distinctIp)
+      updateDistinctIp(document,disCips)
     } else {
       document = new BaseEdgeDocument()
       document.setKey(key)
@@ -151,8 +150,8 @@ object UpdateDocument {
       document.setTo("IP/" + serverIp)
       document.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
       document.addAttribute("LAST_FOUND_TIME", lastFoundTime)
+      putDistinctIp(document,disCips)
       putProtocolAttritube(document, sepAttritubeMap)
-      putDistinctIp(document, distinctIp)
     }
     document
   }

ip-learning-spark/src/test/scala/cn/ac/iie/service/update/UpdateDocumentTest.scala

@@ -1,35 +0,0 @@
-package cn.ac.iie.service.update
-
-import java.util
-import java.util.ArrayList
-import java.util.concurrent.ConcurrentHashMap
-
-import cn.ac.iie.dao.BaseArangoData
-import cn.ac.iie.dao.BaseArangoData._
-import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
-
-import scala.collection.mutable.WrappedArray.ofRef
-
-object UpdateDocumentTest {
-  def main(args: Array[String]): Unit = {
-    val baseArangoData = new BaseArangoData()
-    baseArangoData.readHistoryData("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, classOf[BaseEdgeDocument])
-
-    val value = BaseArangoData.historyRelationFqdnAddressIpMap.keys()
-    while (value.hasMoreElements) {
-      val integer: Integer = value.nextElement()
-      val map: ConcurrentHashMap[String, BaseEdgeDocument] = historyRelationFqdnAddressIpMap.get(integer)
-      val unit = map.keys()
-      while (unit.hasMoreElements) {
-        val key = unit.nextElement()
-        val edgeDocument = map.get(key)
-        //        val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[util.ArrayList[Long]]
-        //        val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[util.ArrayList[String]]
-        val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[Array[String]]
-        val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[Array[java.lang.Long]]
-        println(longs.toString + "---" + strings.toString)
-      }
-    }
-  }
-
-}