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

IP-learning-graph/src/main/java/cn/ac/iie/config/ApplicationConfig.java

@@ -14,7 +14,7 @@ public class ApplicationConfig {
     public static final Integer ARANGODB_BATCH = ConfigUtils.getIntProperty( "arangoDB.batch");
 
     public static final Integer UPDATE_ARANGO_BATCH = ConfigUtils.getIntProperty("update.arango.batch");
-    public static final Long ARANGODB_READ_LIMIT = ConfigUtils.getLongProperty("arangoDB.read.limit");
+    public static final String ARANGODB_READ_LIMIT = ConfigUtils.getStringProperty("arangoDB.read.limit");
 
     public static final Integer THREAD_POOL_NUMBER = ConfigUtils.getIntProperty( "thread.pool.number");
     public static final Integer THREAD_AWAIT_TERMINATION_TIME = ConfigUtils.getIntProperty( "thread.await.termination.time");

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

@@ -42,10 +42,10 @@ public class BaseArangoData {
             }
             CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
 //            long[] timeRange = getTimeRange(table);
-            Long countTotal = getCountTotal(table);
+            Long total = getCountTotal(table);
             for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
 //                String sql = getQuerySql(timeRange, i, table);
-                String sql = getQuerySql(countTotal, i, table);
+                String sql = getQuerySql(total, i, table);
                 ReadHistoryArangoData<T> readHistoryArangoData =
                         new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
                 threadPool.executor(readHistoryArangoData);
@@ -78,12 +78,42 @@ public class BaseArangoData {
     private String getQuerySql(Long cnt,int threadNumber, String table){
         long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER + 1;
         long offsetNum = threadNumber * sepNum;
-        if (sepNum >= ApplicationConfig.ARANGODB_READ_LIMIT * 10000){
-            sepNum = ApplicationConfig.ARANGODB_READ_LIMIT * 10000;
-        }
         return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
     }
 
 
+    private long[] getTimeRange(String table){
+        long minTime = 0L;
+        long maxTime = 0L;
+        long startTime = System.currentTimeMillis();
+        String sql = "LET doc = (FOR doc IN "+table+" RETURN doc) return {max_time:MAX(doc[*].FIRST_FOUND_TIME),min_time:MIN(doc[*].FIRST_FOUND_TIME)}";
+        ArangoCursor<BaseDocument> timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class);
+        try {
+            if (timeDoc != null){
+                while (timeDoc.hasNext()) {
+                    BaseDocument doc = timeDoc.next();
+                    maxTime = Long.parseLong(doc.getAttribute("max_time").toString()) + ApplicationConfig.THREAD_POOL_NUMBER;
+                    minTime = Long.parseLong(doc.getAttribute("min_time").toString());
+                }
+                long lastTime = System.currentTimeMillis();
+                LOG.info(sql+"\n查询最大最小时间用时：" + (lastTime - startTime));
+            }else {
+                LOG.warn("获取ArangoDb时间范围为空");
+            }
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+        return new long[]{minTime, maxTime};
+
+    }
+
+    private String getQuerySql(long[] timeRange,int threadNumber,String table){
+        long minTime = timeRange[0];
+        long maxTime = timeRange[1];
+        long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER;
+        long maxThreadTime = minTime + (threadNumber + 1)* diffTime;
+        long minThreadTime = minTime + threadNumber * diffTime;
+        return "FOR doc IN "+table+" filter doc.FIRST_FOUND_TIME >= "+minThreadTime+" and doc.FIRST_FOUND_TIME <= "+maxThreadTime+" " + ApplicationConfig.ARANGODB_READ_LIMIT + " RETURN doc";
+    }
 
 }

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

@@ -47,11 +47,11 @@ public class UpdateGraphData {
 //            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

@@ -66,11 +66,9 @@ public class ReadClickhouseData {
             long bytesSum = resultSet.getLong("BYTES_SUM");
             String ipType = resultSet.getString("ip_type");
             String[] commonLinkInfos = (String[]) resultSet.getArray("common_link_info").getArray();
-            String commonLinkInfo;
-            if (commonLinkInfos.length > 1 && !commonLinkInfos[1].equals("")){
+            String commonLinkInfo = "";
+            if (commonLinkInfos.length > 1){
                 commonLinkInfo = commonLinkInfos[1];
-            }else {
-                commonLinkInfo = commonLinkInfos[0];
             }
             newDoc.setKey(ip);
             newDoc.addAttribute("IP", ip);
@@ -265,17 +263,16 @@ public class ReadClickhouseData {
     }
 
     public static String getVertexIpSql() {
-//        String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_schema_type != 'BASE'";
-        String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime ;
-        String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
-        String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
+        String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
+        String clientIpSql = "SELECT common_client_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
+        String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
         return "SELECT * FROM((" + clientIpSql + ") UNION ALL (" + serverIpSql + "))";
     }
 
     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

@@ -8,7 +8,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
@@ -59,7 +58,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                     switch (table) {
                         case "R_LOCATE_FQDN2IP":
                             updateProtocolDocument(doc);
-//                            deleteDistinctClientIpByTime(doc);
+                            deleteDistinctClientIpByTime(doc);
                             list.add(doc);
                             break;
                         default:
@@ -74,7 +73,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                 LOG.info(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
             }
         } catch (Exception e) {
-            LOG.error(Arrays.toString(e.getStackTrace()));
+            e.printStackTrace();
         } finally {
             countDownLatch.countDown();
             LOG.info("本线程读取完毕，剩余线程数量：" + countDownLatch.getCount());

IP-learning-graph/src/main/java/cn/ac/iie/service/update/relationship/LocateFqdn2Ip.java

@@ -53,12 +53,12 @@ public class LocateFqdn2Ip extends Relationship {
     }
 
     private void updateDistinctClientIp(BaseEdgeDocument newEdgeDocument,BaseEdgeDocument edgeDocument){
-        ArrayList<String> distCip = (ArrayList<String>) edgeDocument.getAttribute("DIST_CIP");
-        ArrayList<Long> distCipTs = (ArrayList<Long>) edgeDocument.getAttribute("DIST_CIP_TS");
+        String[] distCip = (String[]) edgeDocument.getAttribute("DIST_CIP");
+        long[] distCipTs = (long[]) edgeDocument.getAttribute("DIST_CIP_TS");
         HashMap<String, Long> distCipToTs = new HashMap<>();
-        if (distCip.size() == distCipTs.size()){
-            for (int i = 0;i < distCip.size();i++){
-                distCipToTs.put(distCip.get(i),distCipTs.get(i));
+        if (distCip.length == distCipTs.length){
+            for (int i = 0;i < distCip.length;i++){
+                distCipToTs.put(distCip[i],distCipTs[i]);
             }
         }
         Object[] distCipRecent = (Object[])newEdgeDocument.getAttribute("DIST_CIP");

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

@@ -1,25 +1,27 @@
 #arangoDB参数配置
-arangoDB.host=192.168.44.12
+arangoDB.host=192.168.40.182
+#arangoDB.host=192.168.40.224
 arangoDB.port=8529
 arangoDB.user=upsert
 arangoDB.password=ceiec2018
-arangoDB.DB.name=tsg_galaxy_v3_test
+arangoDB.DB.name=ip-learning-test-0
+#arangoDB.DB.name=tsg_galaxy_v3
 arangoDB.batch=100000
 arangoDB.ttl=3600
 
-arangoDB.read.limit=10
+arangoDB.read.limit=
 update.arango.batch=10000
 
-thread.pool.number=40
+thread.pool.number=80
 thread.await.termination.time=10
 
 
 #读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
 time.limit.type=1
-read.clickhouse.max.time=1603421554
-read.clickhouse.min.time=1603354682
+read.clickhouse.max.time=1598246519
+read.clickhouse.min.time=1597161600
 
 update.interval=3600
-distinct.client.ip.num=100
+distinct.client.ip.num=1
 recent.count.hour=24
 

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

@@ -1,7 +1,9 @@
 drivers=ru.yandex.clickhouse.ClickHouseDriver
 mdb.user=default
-db.id=192.168.44.10:8123/tsg_galaxy_v3?socket_timeout=3600000&compress=0
-mdb.password=ceiec2019
+db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
+mdb.password=111111
+#db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=300000
+#mdb.password=ceiec2019
 initialsize=1
 minidle=1
 maxactive=50

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

@@ -17,6 +17,7 @@ log4j.appender.file.encoding=UTF-8
 log4j.appender.file.Append=true
 #·<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><D8B2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6>Ŀ<EFBFBD><C4BF>
 #log4j.appender.file.file=/home/ceiec/iplearning/logs/ip-learning-application.log
+#log4j.appender.file.file=/home/ceiec/iplearning/testLog/ip-learning-application.log
 log4j.appender.file.file=./logs/ip-learning-application.log
 log4j.appender.file.DatePattern='.'yyyy-MM-dd
 log4j.appender.file.layout=org.apache.log4j.PatternLayout

IP-learning-graph/src/test/java/cn/ac/iie/readHistoryDataTest.java

@@ -6,6 +6,5 @@ import com.arangodb.entity.BaseEdgeDocument;
 public class readHistoryDataTest {
     public static void main(String[] args) {
         BaseArangoData baseArangoData = new BaseArangoData();
-
     }
 }

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

@@ -6,6 +6,7 @@ import cn.ac.iie.utils.ArangoDBConnect;
 import cn.ac.iie.utils.ExecutorThreadPool;
 import com.arangodb.ArangoCursor;
 import com.arangodb.entity.BaseDocument;
+import com.arangodb.entity.BaseEdgeDocument;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,6 +20,15 @@ import java.util.concurrent.CountDownLatch;
  */
 public class BaseArangoData {
     private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
+
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexFqdnMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexIpMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexSubscriberMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnAddressIpMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationIpVisitFqdnMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnSameFqdnMap = new ConcurrentHashMap<>();
+    public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>();
+
     private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
 
     private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
@@ -32,9 +42,11 @@ public class BaseArangoData {
                 historyMap.put(i, new ConcurrentHashMap<>());
             }
             CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER());
-            Long countTotal = getCountTotal(table);
+//            long[] timeRange = getTimeRange(table);
+            Long total = getCountTotal(table);
             for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
-                String sql = getQuerySql(countTotal, 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);
             }
@@ -67,11 +79,50 @@ public class BaseArangoData {
     private String getQuerySql(Long cnt,int threadNumber, String table){
         long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER() + 1;
         long offsetNum = threadNumber * sepNum;
-        if (sepNum >= ApplicationConfig.ARANGODB_READ_LIMIT() * 10000){
-            sepNum = ApplicationConfig.ARANGODB_READ_LIMIT() * 10000;
-        }
         return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
     }
 
+    private long[] getTimeRange(String table) {
+        long minTime = 0L;
+        long maxTime = 0L;
+        long startTime = System.currentTimeMillis();
+        String sql = "LET doc = (FOR doc IN " + table + " RETURN doc) return {max_time:MAX(doc[*].FIRST_FOUND_TIME),min_time:MIN(doc[*].FIRST_FOUND_TIME)}";
+        switch (ApplicationConfig.ARANGO_TIME_LIMIT_TYPE()) {
+            case 0:
+                ArangoCursor<BaseDocument> timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class);
+                try {
+                    if (timeDoc != null) {
+                        while (timeDoc.hasNext()) {
+                            BaseDocument doc = timeDoc.next();
+                            maxTime = Long.parseLong(doc.getAttribute("max_time").toString()) + ApplicationConfig.THREAD_POOL_NUMBER();
+                            minTime = Long.parseLong(doc.getAttribute("min_time").toString());
+                        }
+                    } else {
+                        LOG.warn("获取ArangoDb时间范围为空");
+                    }
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+                break;
+            case 1:
+                maxTime = ApplicationConfig.READ_ARANGO_MAX_TIME();
+                minTime = ApplicationConfig.READ_ARANGO_MIN_TIME();
+                break;
+            default:
+        }
+        long lastTime = System.currentTimeMillis();
+        LOG.warn(sql + "\n查询最大最小时间用时：" + (lastTime - startTime));
+        return new long[]{minTime, maxTime};
+
+    }
+
+    private String getQuerySql(long[] timeRange, int threadNumber, String table) {
+        long minTime = timeRange[0];
+        long maxTime = timeRange[1];
+        long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER();
+        long maxThreadTime = minTime + (threadNumber + 1) * diffTime;
+        long minThreadTime = minTime + threadNumber * diffTime;
+        return "FOR doc IN " + table + " filter doc.FIRST_FOUND_TIME >= " + minThreadTime + " and doc.FIRST_FOUND_TIME <= " + maxThreadTime + " " + ApplicationConfig.ARANGODB_READ_LIMIT() + " RETURN doc";
+    }
 
 }

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

@@ -19,13 +19,12 @@ import java.util.concurrent.CountDownLatch;
  * @author wlh
  * 多线程全量读取arangoDb历史数据，封装到map
  */
-@SuppressWarnings("unchecked")
 public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
     public static long currentHour = System.currentTimeMillis() / (60 * 60 * 1000) * 60 * 60;
     private static final Logger LOG = LoggerFactory.getLogger(ReadHistoryArangoData.class);
-    private static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR();
+    static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR();
 
-    private static final HashSet<String> PROTOCOL_SET;
+    public static final HashSet<String> PROTOCOL_SET;
 
     static {
         PROTOCOL_SET = new HashSet<>();
@@ -59,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();
@@ -70,6 +70,9 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
                             updateProtocolDocument(doc);
                             deleteDistinctClientIpByTime(doc);
                             break;
+//                        case "R_VISIT_IP2FQDN":
+//                            updateProtocolDocument(doc);
+//                            break;
                         default:
                     }
                     int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER();
@@ -93,7 +96,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
             for (String protocol : PROTOCOL_SET) {
                 String protocolRecent = protocol + "_CNT_RECENT";
                 ArrayList<Long> cntRecent = (ArrayList<Long>) doc.getAttribute(protocolRecent);
-                Long[] cntRecentsSrc = cntRecent.toArray(new Long[0]);
+                Long[] cntRecentsSrc = cntRecent.toArray(new Long[cntRecent.size()]);
                 Long[] cntRecentsDst = new Long[RECENT_COUNT_HOUR];
                 System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1);
                 cntRecentsDst[0] = 0L;
@@ -110,6 +113,7 @@ public class ReadHistoryArangoData<T extends BaseDocument> extends Thread {
             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/java/cn/ac/iie/utils/ArangoDBConnect.java

@@ -27,27 +27,14 @@ public class ArangoDBConnect {
     }
 
     private static void getArangoDatabase(){
-        ArangoDB.Builder host = getArangoHost();
-        arangoDB = host
+        arangoDB = new ArangoDB.Builder()
                 .maxConnections(ApplicationConfig.THREAD_POOL_NUMBER())
-                .acquireHostList(true)
+                .host(ApplicationConfig.ARANGODB_HOST(), ApplicationConfig.ARANGODB_PORT())
                 .user(ApplicationConfig.ARANGODB_USER())
                 .password(ApplicationConfig.ARANGODB_PASSWORD())
                 .build();
     }
 
-    private static ArangoDB.Builder getArangoHost(){
-        String hostList = ApplicationConfig.ARANGODB_HOST();
-        String[] split = hostList.split(",");
-        ArangoDB.Builder host = new ArangoDB.Builder();
-        for (String hostStr : split) {
-            host.host(hostStr, ApplicationConfig.ARANGODB_PORT());
-            LOG.warn("arangoDB host {} 已添加",hostStr);
-        }
-        LOG.warn("获取arangoDB host成功");
-        return host;
-    }
-
     public static synchronized ArangoDBConnect getInstance(){
         if (null == conn){
             conn = new ArangoDBConnect();
@@ -83,6 +70,26 @@ public class ArangoDBConnect {
         }
     }
 
+    @Deprecated
+    public <T> void insertAndUpdate(ArrayList<T> docInsert,ArrayList<T> docUpdate,String collectionName){
+        ArangoDatabase database = getDatabase();
+        try {
+            ArangoCollection collection = database.collection(collectionName);
+            if (!docInsert.isEmpty()){
+                collection.importDocuments(docInsert);
+            }
+            if (!docUpdate.isEmpty()){
+                collection.replaceDocuments(docUpdate);
+            }
+        }catch (Exception e){
+            System.out.println("更新失败");
+            e.printStackTrace();
+        }finally {
+            docInsert.clear();
+            docInsert.clear();
+        }
+    }
+
     public <T> void overwrite(ArrayList<T> docOverwrite,String collectionName){
         ArangoDatabase database = getDatabase();
         try {
@@ -94,11 +101,11 @@ public class ArangoDBConnect {
                 MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
                 Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
                 for (ErrorEntity errorEntity:errors){
-                    LOG.debug("写入arangoDB异常："+errorEntity.getErrorMessage());
+                    LOG.warn("写入arangoDB异常："+errorEntity.getErrorMessage());
                 }
             }
         }catch (Exception e){
-            LOG.error("更新arangoDB失败："+e.toString());
+            System.out.println("更新失败："+e.toString());
         }finally {
             docOverwrite.clear();
         }

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

@@ -1,40 +1,48 @@
 #spark任务配置
-spark.sql.shuffle.partitions=10
+spark.sql.shuffle.partitions=5
 spark.executor.memory=4g
 spark.app.name=test
 spark.network.timeout=300s
+repartitionNumber=36
 spark.serializer=org.apache.spark.serializer.KryoSerializer
-#spark.serializer=org.apache.spark.serializer.JavaSerializer
 master=local[*]
 #spark读取clickhouse配置
-spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.10: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=ceiec2019
-spark.read.clickhouse.numPartitions=5
+spark.read.clickhouse.numPartitions=144
 spark.read.clickhouse.fetchsize=10000
-spark.read.clickhouse.partitionColumn=LAST_FOUND_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.123,192.168.40.223,192.168.40.222
-#arangoDB.host=192.168.40.223
+arangoDB.host=192.168.40.182
 arangoDB.port=8529
 arangoDB.user=upsert
-arangoDB.password=ceiec2019
-arangoDB.DB.name=tsg_galaxy_v3
+arangoDB.password=ceiec2018
+#arangoDB.DB.name=insert_iplearn_index
+arangoDB.DB.name=ip-learning-test-0
 arangoDB.ttl=3600
 
-thread.pool.number=10
+thread.pool.number=5
 
 #读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
 clickhouse.time.limit.type=1
-read.clickhouse.max.time=1600916194
-read.clickhouse.min.time=1599197648
+read.clickhouse.max.time=1598246519
+read.clickhouse.min.time=1597161600
 
-arangoDB.read.sepNum=10
+#读取arangoDB时间范围方式，0：正常读；1：指定时间范围
+arango.time.limit.type=0
+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=3600

ip-learning-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala

@@ -36,7 +36,12 @@ object ApplicationConfig {
   val READ_CLICKHOUSE_MAX_TIME: Long = config.getLong("read.clickhouse.max.time")
   val READ_CLICKHOUSE_MIN_TIME: Long = config.getLong("read.clickhouse.min.time")
 
-  val ARANGODB_READ_LIMIT: Long = config.getLong("arangoDB.read.sepNum")
+  val ARANGO_TIME_LIMIT_TYPE: Int = config.getInt("arango.time.limit.type")
+
+  val READ_ARANGO_MAX_TIME: Long = config.getLong("read.arango.max.time")
+  val READ_ARANGO_MIN_TIME: Long = config.getLong("read.arango.min.time")
+
+  val ARANGODB_READ_LIMIT: String = config.getString("arangoDB.read.limit")
   val UPDATE_ARANGO_BATCH: Int = config.getInt("update.arango.batch")
   val RECENT_COUNT_HOUR: Int = config.getInt("recent.count.hour")
   val DISTINCT_CLIENT_IP_NUM: Int = config.getInt("distinct.client.ip.num")

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

@@ -28,16 +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 + " AND common_schema_type != 'BASE'"
+    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
@@ -47,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"""
@@ -107,35 +128,45 @@ object BaseClickhouseData {
 
   def getVertexIpDf: DataFrame ={
     loadConnectionDataFromCk()
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
     val sql =
-      s"""
-        |(SELECT * FROM
-        |((SELECT common_client_ip AS IP,MIN(common_end_time) AS FIRST_FOUND_TIME,
-        |MAX(common_end_time) AS LAST_FOUND_TIME,
-        |count(*) as SESSION_COUNT,
-        |SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
-        |groupUniqArray(2)(common_link_info_c2s)[2] as common_link_info,
-        |'client' as ip_type
-        |FROM tsg_galaxy_v3.connection_record_log
-        |where $where
-        |group by common_client_ip)
-        |UNION ALL
-        |(SELECT common_server_ip AS IP,
-        |MIN(common_end_time) AS FIRST_FOUND_TIME,
-        |MAX(common_end_time) AS LAST_FOUND_TIME,
-        |count(*) as SESSION_COUNT,
-        |SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
-        |groupUniqArray(2)(common_link_info_s2c)[2] as common_link_info,
-        |'server' as ip_type
-        |FROM tsg_galaxy_v3.connection_record_log
-        |where $where
-        |group by common_server_ip))) as dbtable
+      """
+        |SELECT
+        | *
+        |FROM
+        | (
+        |  (
+        |   SELECT
+        |    common_client_ip AS IP,
+        |    MIN(common_recv_time) AS FIRST_FOUND_TIME,
+        |    MAX(common_recv_time) AS LAST_FOUND_TIME,
+        |    count(*) as SESSION_COUNT,
+        |    sum(common_c2s_byte_num) as BYTES_SUM,
+        |    'client' as ip_type
+        |   FROM
+        |    global_temp.dbtable
+        |   GROUP BY
+        |    IP
+        |  )
+        | UNION ALL
+        |  (
+        |   SELECT
+        |    common_server_ip AS IP,
+        |    MIN(common_recv_time) AS FIRST_FOUND_TIME,
+        |    MAX(common_recv_time) AS LAST_FOUND_TIME,
+        |    count(*) as SESSION_COUNT,
+        |    sum(common_s2c_byte_num) as BYTES_SUM,
+        |    'server' as ip_type
+        |   FROM
+        |    global_temp.dbtable
+        |   GROUP BY
+        |    IP
+        |  )
+        | )
       """.stripMargin
     LOG.warn(sql)
-    val frame = initClickhouseData(sql)
-    frame.printSchema()
-    frame
+    val vertexIpDf = spark.sql(sql)
+    vertexIpDf.printSchema()
+    vertexIpDf
   }
 
   /*
@@ -146,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
@@ -164,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
@@ -185,91 +214,6 @@ object BaseClickhouseData {
   }
   */
 
-  def getRelationFqdnLocateIpDf: DataFrame ={
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
-    val sql =
-      s"""
-         |(SELECT * FROM
-         |((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,
-         |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_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_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
-  }
-
-  def getRelationSubidLocateIpDf: DataFrame ={
-    val where =
-      s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
-         | AND common_subscriber_id != ''
-         | AND radius_framed_ip != ''
-      """.stripMargin
-    val sql =
-      s"""
-         |(
-         |SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME
-         |FROM radius_record_log
-         |WHERE $where GROUP BY common_subscriber_id,radius_framed_ip
-         |) as dbtable
-      """.stripMargin
-    LOG.warn(sql)
-    val frame = initClickhouseData(sql)
-    frame.printSchema()
-    frame
-  }
-
-  def getVertexSubidDf: DataFrame ={
-    val where =
-      s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
-         | AND common_subscriber_id != ''
-         | AND radius_framed_ip != ''
-      """.stripMargin
-    val sql =
-      s"""
-        |(
-        |SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM radius_record_log
-        |WHERE $where GROUP BY common_subscriber_id
-        |)as dbtable
-      """.stripMargin
-    LOG.warn(sql)
-    val frame = initClickhouseData(sql)
-    frame.printSchema()
-    frame
-  }
-
-  def getVertexFramedIpDf: DataFrame ={
-    val where =
-      s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
-         | AND common_subscriber_id != ''
-         | AND radius_framed_ip != ''
-      """.stripMargin
-    val sql =
-      s"""
-        |(
-        |SELECT DISTINCT radius_framed_ip,common_recv_time as LAST_FOUND_TIME FROM radius_record_log WHERE $where
-        |)as dbtable
-      """.stripMargin
-    LOG.warn(sql)
-    val frame = initClickhouseData(sql)
-    frame.printSchema()
-    frame
-  }
-
   private def getTimeLimit: (Long,Long) ={
     var maxTime = 0L
     var minTime = 0L

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

@@ -20,17 +20,6 @@ object MergeDataFrame {
       .partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
   }
 
-  def mergeVertexFrameIp: RDD[Row] ={
-    val values = BaseClickhouseData.getVertexFramedIpDf
-      .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
-      .rdd.map(row => {
-      val ip = row.getAs[String]("radius_framed_ip")
-      (ip, row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
-    LOG.warn(s"读取R_LOCATE_SUBSCRIBER2IP clickhouse成功，共：${values.count()} 条")
-    values
-  }
-
   def mergeVertexIp(): RDD[Row]={
     val vertexIpDf = BaseClickhouseData.getVertexIpDf
     val frame = vertexIpDf.groupBy("IP").agg(
@@ -38,8 +27,7 @@ object MergeDataFrame {
       max("LAST_FOUND_TIME").alias("LAST_FOUND_TIME"),
       collect_list("SESSION_COUNT").alias("SESSION_COUNT_LIST"),
       collect_list("BYTES_SUM").alias("BYTES_SUM_LIST"),
-      collect_list("ip_type").alias("ip_type_list"),
-      last("common_link_info").alias("common_link_info")
+      collect_list("ip_type").alias("ip_type_list")
     )
     val values = frame.rdd.map(row => (row.get(0), row))
       .partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
@@ -47,9 +35,7 @@ object MergeDataFrame {
   }
 
   def mergeRelationFqdnLocateIp(): RDD[Row] ={
-    val frame = BaseClickhouseData.getRelationFqdnLocateIpDf
-      .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
-      .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,50 +44,23 @@ object MergeDataFrame {
         collect_list("schema_type").alias("schema_type_list"),
         collect_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT")
       )
-    val values = frame.rdd.map(row => {
+    frame.rdd.map(row => {
       val fqdn = row.getAs[String]("FQDN")
       val serverIp = row.getAs[String]("common_server_ip")
-      val key = fqdn.concat("-" + serverIp)
-      (key, row)
+      val key = fqdn.concat("-"+serverIp)
+      (key,row)
     }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
-    LOG.warn(s"读取R_LOCATE_FQDN2IP clickhouse成功，共：${values.count()} 条")
-    values
 
   }
 
-  def mergeRelationSubidLocateIp(): RDD[Row] ={
-    val values = BaseClickhouseData.getRelationSubidLocateIpDf
-      .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
-      .rdd.map(row => {
-      val commonSubscriberId = row.getAs[String]("common_subscriber_id")
-      val ip = row.getAs[String]("radius_framed_ip")
-      val key = commonSubscriberId.concat("-" + ip)
-      (key, row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
-    LOG.warn(s"读取R_LOCATE_SUBSCRIBER2IP clickhouse成功，共：${values.count()} 条")
-    values
-  }
-
-  def mergeVertexSubid(): RDD[Row] ={
-    val values = BaseClickhouseData.getVertexSubidDf
-      .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
-      .rdd.map(row => {
-      val commonSubscriberId = row.getAs[String]("common_subscriber_id")
-      (commonSubscriberId, row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
-    LOG.warn(s"读取SUBSCRIBER clickhouse成功，共：${values.count()} 条")
-    values
-  }
-
   private def isDomain(fqdn: String): Boolean = {
     try {
       if (fqdn == null || fqdn.length == 0) {
         return false
       }
-
-      val fqdnArr = fqdn.split(":")(0).split("\\.")
-
-      if (fqdnArr.length != 4){
+      val domain = fqdn.split(":")(0)
+      val fqdnArr = domain.split("\\.")
+      if (fqdnArr.length < 4 || fqdnArr.length > 4){
         return true
       }
       for (f <- fqdnArr) {
@@ -116,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

@@ -26,10 +26,6 @@ object UpdateDocHandler {
     hisDoc.addAttribute(attributeName,newAttribute+hisAttritube)
   }
 
-  def replaceAttribute(hisDoc: BaseDocument,newAttribute:String,attributeName:String): Unit ={
-    hisDoc.addAttribute(attributeName,newAttribute)
-  }
-
   def separateAttributeByIpType(ipTypeList:ofRef[String],
                                 sessionCountList:ofRef[AnyRef],
                                 bytesSumList:ofRef[AnyRef]): (Long,Long,Long,Long) ={
@@ -98,12 +94,13 @@ object UpdateDocHandler {
   }
 
   def mergeDistinctIp(distCipRecent:ofRef[String]): Array[String] ={
-     distCipRecent.flatMap(str => {
-       str.replaceAll("\\[","")
-       .replaceAll("\\]","")
-       .replaceAll("\\'","")
-         .split(",")
-     }).distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
+    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

@@ -5,6 +5,7 @@ import java.util.concurrent.ConcurrentHashMap
 
 import cn.ac.iie.config.ApplicationConfig
 import cn.ac.iie.dao.BaseArangoData
+import cn.ac.iie.dao.BaseArangoData._
 import cn.ac.iie.service.transform.MergeDataFrame._
 import cn.ac.iie.service.update.UpdateDocHandler._
 import cn.ac.iie.utils.{ArangoDBConnect, ExecutorThreadPool, SparkSessionUtil}
@@ -25,12 +26,7 @@ object UpdateDocument {
 
   def update(): Unit = {
     try {
-//      updateDocument("FQDN", getVertexFqdnRow, classOf[BaseDocument], mergeVertexFqdn)
-//      updateDocument("IP", getVertexIpRow, classOf[BaseDocument], mergeVertexIp)
       updateDocument("R_LOCATE_FQDN2IP", getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
-      updateDocument("SUBSCRIBER",getVertexSubidRow,classOf[BaseDocument],mergeVertexSubid)
-      insertFrameIp()
-      updateDocument("R_LOCATE_SUBSCRIBER2IP",getRelationSubidLocateIpRow,classOf[BaseEdgeDocument],mergeRelationSubidLocateIp)
     } catch {
       case e: Exception => e.printStackTrace()
     } finally {
@@ -40,27 +36,6 @@ object UpdateDocument {
     }
   }
 
-  private def insertFrameIp(): Unit ={
-    mergeVertexFrameIp.foreachPartition(iter => {
-      val resultDocumentList = new util.ArrayList[BaseDocument]
-      var i = 0
-      iter.foreach(row => {
-        val document = getVertexFrameipRow(row)
-        resultDocumentList.add(document)
-        i += 1
-        if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
-          arangoManger.overwrite(resultDocumentList, "IP")
-          LOG.warn(s"更新:IP" + i)
-          i = 0
-        }
-      })
-      if (i != 0) {
-        arangoManger.overwrite(resultDocumentList, "IP")
-        LOG.warn(s"更新IP:" + i)
-      }
-    })
-  }
-
   private def updateDocument[T <: BaseDocument](collName: String,
                                                 getDocumentRow: (Row, ConcurrentHashMap[String, T]) => T,
                                                 clazz: Class[T],
@@ -68,6 +43,7 @@ object UpdateDocument {
                                                ): Unit = {
     val historyMap = baseArangoData.readHistoryData(collName, clazz)
     val hisBc = spark.sparkContext.broadcast(historyMap)
+    LOG.warn("广播变量发送完毕")
     try {
       val start = System.currentTimeMillis()
       val newDataRdd = getNewDataRdd()
@@ -117,56 +93,6 @@ object UpdateDocument {
     document
   }
 
-  private def getRelationSubidLocateIpRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseEdgeDocument]): BaseEdgeDocument ={
-    val subId = row.getAs[String]("common_subscriber_id")
-    val ip = row.getAs[String]("radius_framed_ip")
-    val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
-    val firstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
-
-    val key = subId.concat("-"+ip)
-    var document = dictionaryMap.getOrDefault(key,null)
-    if (document != null){
-      updateMaxAttribute(document,lastFoundTime,"LAST_FOUND_TIME")
-    } else {
-      document = new BaseEdgeDocument()
-      document.setKey(key)
-      document.setFrom("SUBSCRIBER/" + subId)
-      document.setTo("IP/" + ip)
-      document.addAttribute("SUBSCRIBER",subId)
-      document.addAttribute("IP",ip)
-      document.addAttribute("FIRST_FOUND_TIME",firstFoundTime)
-      document.addAttribute("LAST_FOUND_TIME",lastFoundTime)
-    }
-
-    document
-  }
-
-  private def getVertexFrameipRow(row: Row): BaseDocument ={
-    val ip = row.getAs[String]("radius_framed_ip")
-    val document = new BaseDocument()
-    document.setKey(ip)
-    document.addAttribute("IP",ip)
-    document
-  }
-
-  private def getVertexSubidRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseDocument]): BaseDocument ={
-    val subId = row.getAs[String]("common_subscriber_id")
-    val subLastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
-    val subFirstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
-    var document = dictionaryMap.getOrDefault(subId,null)
-    if (document != null){
-      updateMaxAttribute(document,subLastFoundTime,"LAST_FOUND_TIME")
-    } else {
-      document = new BaseDocument()
-      document.setKey(subId)
-      document.addAttribute("SUBSCRIBER",subId)
-      document.addAttribute("FIRST_FOUND_TIME",subFirstFoundTime)
-      document.addAttribute("LAST_FOUND_TIME",subLastFoundTime)
-    }
-
-    document
-  }
-
   private def getVertexIpRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseDocument]): BaseDocument = {
     val ip = row.getAs[String]("IP")
     val firstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
@@ -174,7 +100,6 @@ object UpdateDocument {
     val sessionCountList = row.getAs[ofRef[AnyRef]]("SESSION_COUNT_LIST")
     val bytesSumList = row.getAs[ofRef[AnyRef]]("BYTES_SUM_LIST")
     val ipTypeList = row.getAs[ofRef[String]]("ip_type_list")
-    val linkInfo = row.getAs[String]("common_link_info")
     val sepAttributeTuple = separateAttributeByIpType(ipTypeList, sessionCountList, bytesSumList)
 
     var document = dictionaryMap.getOrDefault(ip, null)
@@ -184,7 +109,6 @@ object UpdateDocument {
       updateSumAttribute(document, sepAttributeTuple._2, "SERVER_BYTES_SUM")
       updateSumAttribute(document, sepAttributeTuple._3, "CLIENT_SESSION_COUNT")
       updateSumAttribute(document, sepAttributeTuple._4, "CLIENT_BYTES_SUM")
-      replaceAttribute(document,linkInfo,"COMMON_LINK_INFO")
     } else {
       document = new BaseDocument
       document.setKey(ip)
@@ -195,7 +119,7 @@ object UpdateDocument {
       document.addAttribute("SERVER_BYTES_SUM", sepAttributeTuple._2)
       document.addAttribute("CLIENT_SESSION_COUNT", sepAttributeTuple._3)
       document.addAttribute("CLIENT_BYTES_SUM", sepAttributeTuple._4)
-      document.addAttribute("COMMON_LINK_INFO", linkInfo)
+      document.addAttribute("COMMON_LINK_INFO", "")
     }
     document
   }
@@ -207,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[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)
@@ -225,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)
-//      }
-//    }
-//  }
-//
-//}