[修复] 增加iplearning对IP和VSYS_ID的非空判断，修改日志信息中文为英文

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 String ARANGODB_READ_LIMIT = ConfigUtils.getStringProperty("arangoDB.read.limit");
+    public static final Long ARANGODB_READ_LIMIT = ConfigUtils.getLongProperty("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

@@ -129,6 +129,9 @@ 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){
+            sepNum = ApplicationConfig.ARANGODB_READ_LIMIT;
+        }
         return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
     }
 

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

@@ -44,8 +44,8 @@ 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);
@@ -79,8 +79,8 @@ public class UpdateGraphData {
         long start = System.currentTimeMillis();
         try {
 
-            updateDocument("FQDN", Fqdn.class,BaseDocument.class,
-                    ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
+//            updateDocument("FQDN", Fqdn.class,BaseDocument.class,
+//                    ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
 
             updateDocument("IP", Ip.class,BaseDocument.class,
                     ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);

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

@@ -140,11 +140,11 @@ public class ReadClickhouseData {
         try {
             String vFqdn = resultSet.getString("FQDN");
             if (isDomain(vFqdn)) {
-                String vIp = resultSet.getString("common_server_ip");
+                String vIp = resultSet.getString("server_ip");
                 long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
                 long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
                 long countTotal = resultSet.getLong("COUNT_TOTAL");
-                String schemaType = resultSet.getString("schema_type");
+                String schemaType = resultSet.getString("decoded_as");
                 String[] distCipRecents = (String[]) resultSet.getArray("DIST_CIP_RECENT").getArray();
                 long[] clientIpTs = new long[distCipRecents.length];
                 for (int i = 0; i < clientIpTs.length; i++) {
@@ -174,12 +174,12 @@ public class ReadClickhouseData {
         try {
             String vFqdn = resultSet.getString("FQDN");
             if (isDomain(vFqdn)) {
-                String vIp = resultSet.getString("common_client_ip");
+                String vIp = resultSet.getString("client_ip");
                 String key = vIp + "-" + vFqdn;
                 long firstFoundTime = resultSet.getLong("FIRST_FOUND_TIME");
                 long lastFoundTime = resultSet.getLong("LAST_FOUND_TIME");
                 long countTotal = resultSet.getLong("COUNT_TOTAL");
-                String schemaType = resultSet.getString("schema_type");
+                String schemaType = resultSet.getString("decoded_as");
 
                 newDoc = new BaseEdgeDocument();
                 newDoc.setKey(key);
@@ -256,41 +256,41 @@ public class ReadClickhouseData {
     }
 
     public static String getVertexFqdnSql() {
-        String where = "common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
-        String sslSql = "SELECT ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'SSL' GROUP BY ssl_sni";
-        String httpSql = "SELECT http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host";
+        String where = "recv_time >= " + minTime + " AND recv_time < " + maxTime;
+        String sslSql = "SELECT ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.session_record WHERE " + where + " and decoded_as = 'SSL' GROUP BY ssl_sni";
+        String httpSql = "SELECT http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME FROM tsg_galaxy_v3.session_record WHERE " + where + " and decoded_as = 'HTTP' GROUP BY http_host";
         return "SELECT FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME FROM ((" + sslSql + ") UNION ALL (" + httpSql + ")) GROUP BY FQDN HAVING FQDN != ''";
     }
 
     public static String getVertexIpSql() {
-        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";
+        String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
+        String clientIpSql = "SELECT client_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(sent_bytes+received_bytes) as BYTES_SUM,groupUniqArray(2)(common_link_info_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
+        String serverIpSql = "SELECT server_ip AS IP, MIN(recv_time) AS FIRST_FOUND_TIME,MAX(recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(sent_bytes+received_bytes) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.session_record 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 = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
+        String sslSql = "SELECT ssl_sni AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(client_ip) AS DIST_CIP_RECENT,'TLS' AS  decoded_as FROM tsg_galaxy_v3.session_record WHERE " + where + " and decoded_as = 'SSL' GROUP BY ssl_sni,server_ip";
+        String httpSql = "SELECT http_host AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,groupUniqArray("+DISTINCT_CLIENT_IP_NUM+")(client_ip) AS DIST_CIP_RECENT,'HTTP' AS decoded_as FROM tsg_galaxy_v3.session_record WHERE " + where + " and decoded_as = 'HTTP' GROUP BY http_host,server_ip";
         return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
     }
 
     public static String getRelationshipIpVisitFqdnSql() {
-        String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime;
-        String httpSql = "SELECT http_host AS FQDN,common_client_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'HTTP' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_client_ip";
-        String sslSql = "SELECT ssl_sni AS FQDN,common_client_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'TLS' AS schema_type FROM tsg_galaxy_v3.connection_record_log WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
+        String where = " recv_time >= " + minTime + " AND recv_time < " + maxTime;
+        String httpSql = "SELECT http_host AS FQDN,client_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'HTTP' AS decoded_as FROM tsg_galaxy_v3.session_record WHERE " + where + " and decoded_as = 'HTTP' GROUP BY http_host,client_ip";
+        String sslSql = "SELECT ssl_sni AS FQDN,client_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,'TLS' AS decoded_as FROM tsg_galaxy_v3.session_record WHERE decoded_as = 'SSL' GROUP BY ssl_sni,client_ip";
         return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
     }
 
     public static String getVertexSubscriberSql() {
         String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
-        return "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";
+        return "SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM radius_record WHERE" + where + " GROUP BY common_subscriber_id";
     }
 
     public static String getRelationshipSubsciberLocateIpSql() {
         String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_framed_ip != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
-        return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip";
+        return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip";
     }
 
     private static long[] getTimeLimit() {

IP-learning-graph/src/main/java/cn/ac/iie/service/update/vertex/Ip.java

@@ -33,10 +33,10 @@ public class Ip extends Vertex {
     }
 
     private void updateIpByType(BaseDocument newDocument, BaseDocument historyDocument) {
-        putSumAttribute(newDocument, historyDocument, "CLIENT_SESSION_COUNT");
-        putSumAttribute(newDocument, historyDocument, "CLIENT_BYTES_SUM");
-        putSumAttribute(newDocument, historyDocument, "SERVER_SESSION_COUNT");
-        putSumAttribute(newDocument, historyDocument, "SERVER_BYTES_SUM");
+//        putSumAttribute(newDocument, historyDocument, "CLIENT_SESSION_COUNT");
+//        putSumAttribute(newDocument, historyDocument, "CLIENT_BYTES_SUM");
+//        putSumAttribute(newDocument, historyDocument, "SERVER_SESSION_COUNT");
+//        putSumAttribute(newDocument, historyDocument, "SERVER_BYTES_SUM");
     }
 
 }

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

@@ -6,7 +6,6 @@ public class IpLearningApplicationTest {
 
     public static void main(String[] args) {
         UpdateGraphData updateGraphData = new UpdateGraphData();
-//        updateGraphData.updateArango();
         updateGraphData.updateArango2();
 
     }

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

@@ -1,15 +1,13 @@
 #arangoDB参数配置
-arangoDB.host=192.168.40.182
-#arangoDB.host=192.168.40.224
+arangoDB.host=192.168.44.12
 arangoDB.port=8529
-arangoDB.user=upsert
-arangoDB.password=ceiec2018
-arangoDB.DB.name=ip-learning-test
-#arangoDB.DB.name=tsg_galaxy_v3
+arangoDB.user=root
+arangoDB.password=ceiec2019
+arangoDB.DB.name=tsg_galaxy_v3
 arangoDB.batch=100000
 arangoDB.ttl=3600
 
-arangoDB.read.limit=
+arangoDB.read.limit=10000000
 update.arango.batch=10000
 
 thread.pool.number=10
@@ -22,6 +20,6 @@ read.clickhouse.max.time=1596684142
 read.clickhouse.min.time=1596425769
 
 update.interval=3600
-distinct.client.ip.num=10000
+distinct.client.ip.num=1000
 recent.count.hour=24
 

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

@@ -1,9 +1,7 @@
 drivers=ru.yandex.clickhouse.ClickHouseDriver
 mdb.user=default
-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
+db.id=192.168.44.67:8123/tsg_galaxy_v3?socket_timeout=3600000
+mdb.password=ceiec2019
 initialsize=1
 minidle=1
 maxactive=50

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

@@ -1,53 +0,0 @@
-package cn.ac.iie;
-
-import cn.ac.iie.dao.BaseArangoData;
-import cn.ac.iie.utils.ArangoDBConnect;
-import cn.ac.iie.utils.ExecutorThreadPool;
-import com.arangodb.entity.BaseDocument;
-import com.arangodb.entity.BaseEdgeDocument;
-import org.junit.After;
-import org.junit.Test;
-
-import java.util.Enumeration;
-import java.util.concurrent.ConcurrentHashMap;
-
-public class TestReadArango {
-    private static ExecutorThreadPool pool = ExecutorThreadPool.getInstance();
-    private static ArangoDBConnect arangoManger = ArangoDBConnect.getInstance();
-
-    private static BaseArangoData baseArangoData = new BaseArangoData();
-
-
-    @Test
-    public void testReadFqdnFromArango() {
-        ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyData =
-                baseArangoData.readHistoryData("FQDN", BaseDocument.class);
-        printMap(historyData);
-    }
-
-    @Test
-    public void testReadFqdnLocIpFromArango() {
-        ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> ip =
-                baseArangoData.readHistoryData("R_LOCATE_FQDN2IP", BaseEdgeDocument.class);
-        printMap(ip);
-    }
-
-    private <T extends BaseDocument> void printMap(ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyData) {
-        ConcurrentHashMap<String, T> map = historyData.get(2);
-        Enumeration<String> keys = map.keys();
-        while (keys.hasMoreElements()) {
-            String key = keys.nextElement();
-            T document = map.get(key);
-            System.out.println(document.toString());
-        }
-    }
-
-
-    @After
-    public void clearSource() {
-        pool.shutdown();
-        arangoManger.clean();
-    }
-
-
-}

ip-learning-spark/pom.xml

@@ -7,99 +7,263 @@
     <groupId>cn.ac.iie</groupId>
     <artifactId>ip-learning-spark</artifactId>
     <version>1.0-SNAPSHOT</version>
-<dependencies>
-    <dependency>
-        <groupId>javax.servlet</groupId>
-        <artifactId>javax.servlet-api</artifactId>
-        <version>3.0.1</version>
-    </dependency>
 
-    <dependency>
-        <groupId>org.apache.httpcomponents</groupId>
-        <artifactId>httpclient</artifactId>
-        <version>4.5.2</version>
-    </dependency>
+    <repositories>
+        <repository>
+            <id>nexus</id>
+            <name>Team Nexus Repository</name>
+            <url>http://192.168.40.125:8099/content/groups/public</url>
+        </repository>
 
-    <!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpcore -->
-    <dependency>
-        <groupId>org.apache.httpcomponents</groupId>
-        <artifactId>httpcore</artifactId>
-        <version>4.4.6</version>
-    </dependency>
+        <repository>
+            <id>ebi</id>
+            <name>www.ebi.ac.uk</name>
+            <url>http://www.ebi.ac.uk/intact/maven/nexus/content/groups/public/</url>
+        </repository>
 
-    <dependency>
-        <groupId>com.google.guava</groupId>
-        <artifactId>guava</artifactId>
-        <version>19.0</version>
-    </dependency>
+        <repository>
+            <id>maven-ali</id>
+            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
+            <releases>
+                <enabled>true</enabled>
+            </releases>
+            <snapshots>
+                <enabled>true</enabled>
+                <updatePolicy>always</updatePolicy>
+                <checksumPolicy>fail</checksumPolicy>
+            </snapshots>
+        </repository>
 
-    <dependency>
-        <groupId>org.apache.spark</groupId>
-        <artifactId>spark-core_2.11</artifactId>
-        <version>2.2.3</version>
-    </dependency>
+    </repositories>
 
-    <dependency>
-        <groupId>org.apache.spark</groupId>
-        <artifactId>spark-sql_2.11</artifactId>
-        <version>2.2.3</version>
-    </dependency>
+    <dependencies>
 
-    <dependency>
-        <groupId>ru.yandex.clickhouse</groupId>
-        <artifactId>clickhouse-jdbc</artifactId>
-        <version>0.1.54</version>
-    </dependency>
+        <dependency>
+            <groupId>com.zdjizhi</groupId>
+            <artifactId>galaxy</artifactId>
+            <version>1.0.6</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>log4j-over-slf4j</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
 
-    <dependency>
-        <groupId>com.typesafe</groupId>
-        <artifactId>config</artifactId>
-        <version>1.2.1</version>
-    </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.7.25</version>
+        </dependency>
 
-    <dependency>
-        <groupId>com.arangodb</groupId>
-        <artifactId>arangodb-java-driver</artifactId>
-        <version>6.6.3</version>
-    </dependency>
+        <dependency>
+            <groupId>javax.servlet</groupId>
+            <artifactId>javax.servlet-api</artifactId>
+            <version>3.0.1</version>
+        </dependency>
 
-    <dependency>
-        <groupId>com.arangodb</groupId>
-        <artifactId>velocypack-module-jdk8</artifactId>
-        <version>1.1.0</version>
-    </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.2</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>httpcore</artifactId>
+                    <groupId>org.apache.httpcomponents</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
 
-    <dependency>
-        <groupId>com.arangodb</groupId>
-        <artifactId>velocypack-module-scala_2.11</artifactId>
-        <version>1.2.0</version>
-    </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpcore -->
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+            <version>4.4.6</version>
+        </dependency>
 
-    <dependency>
-        <groupId>org.scala-lang</groupId>
-        <artifactId>scala-library</artifactId>
-        <version>2.11.8</version>
-    </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>19.0</version>
+        </dependency>
 
-    <dependency>
-        <groupId>net.alchim31.maven</groupId>
-        <artifactId>scala-maven-plugin</artifactId>
-        <version>3.2.0</version>
-    </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.11</artifactId>
+            <version>2.2.3</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>guava</artifactId>
+                    <groupId>com.google.guava</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>netty</artifactId>
+                    <groupId>io.netty</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>javax.servlet-api</artifactId>
+                    <groupId>javax.servlet</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>jaxb-api</artifactId>
+                    <groupId>javax.xml.bind</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>log4j</artifactId>
+                    <groupId>log4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>httpclient</artifactId>
+                    <groupId>org.apache.httpcomponents</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>httpcore</artifactId>
+                    <groupId>org.apache.httpcomponents</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>zookeeper</artifactId>
+                    <groupId>org.apache.zookeeper</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>slf4j-api</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
 
-    <dependency>
-        <groupId>org.scala-lang.modules</groupId>
-        <artifactId>scala-xml_2.11</artifactId>
-        <version>1.0.4</version>
-    </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.11</artifactId>
+            <version>2.2.3</version>
+        </dependency>
 
-    <dependency>
-        <groupId>org.scala-tools</groupId>
-        <artifactId>maven-scala-plugin</artifactId>
-        <version>2.15.2</version>
-    </dependency>
+        <dependency>
+            <groupId>ru.yandex.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>0.1.54</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>jackson-databind</artifactId>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>slf4j-api</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
 
-</dependencies>
+        <dependency>
+            <groupId>com.typesafe</groupId>
+            <artifactId>config</artifactId>
+            <version>1.2.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.arangodb</groupId>
+            <artifactId>arangodb-java-driver</artifactId>
+            <version>6.6.3</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.arangodb</groupId>
+            <artifactId>velocypack-module-jdk8</artifactId>
+            <version>1.1.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.arangodb</groupId>
+            <artifactId>velocypack-module-scala_2.11</artifactId>
+            <version>1.2.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <version>2.11.8</version>
+        </dependency>
+
+        <dependency>
+            <groupId>net.alchim31.maven</groupId>
+            <artifactId>scala-maven-plugin</artifactId>
+            <version>3.2.0</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>maven-artifact</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>maven-core</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>maven-model</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>maven-plugin-api</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>maven-repository-metadata</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>maven-settings</artifactId>
+                    <groupId>org.apache.maven</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>doxia-sink-api</artifactId>
+                    <groupId>org.apache.maven.doxia</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>plexus-container-default</artifactId>
+                    <groupId>org.codehaus.plexus</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>plexus-interpolation</artifactId>
+                    <groupId>org.codehaus.plexus</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>plexus-utils</artifactId>
+                    <groupId>org.codehaus.plexus</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>scala-library</artifactId>
+                    <groupId>org.scala-lang</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>scala-reflect</artifactId>
+                    <groupId>org.scala-lang</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.scala-lang.modules</groupId>
+            <artifactId>scala-xml_2.11</artifactId>
+            <version>1.0.4</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.scala-tools</groupId>
+            <artifactId>maven-scala-plugin</artifactId>
+            <version>2.15.2</version>
+        </dependency>
+
+    </dependencies>
 
     <build>
         <plugins>

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

@@ -1,75 +0,0 @@
-package cn.ac.iie.dao;
-
-import cn.ac.iie.config.ApplicationConfig;
-import cn.ac.iie.service.read.ReadHistoryArangoData;
-import cn.ac.iie.utils.ArangoDBConnect;
-import cn.ac.iie.utils.ExecutorThreadPool;
-import com.arangodb.ArangoCursor;
-import com.arangodb.entity.BaseDocument;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * 获取arangoDB历史数据
- *
- * @author wlh
- */
-public class BaseArangoData {
-    private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
-    private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
-
-    private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
-
-    public <T extends BaseDocument> void readHistoryData(String table,
-                                                  ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
-                                                  Class<T> type) {
-        try {
-            LOG.warn("开始更新" + table);
-            long start = System.currentTimeMillis();
-            for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
-                historyMap.put(i, new ConcurrentHashMap<>());
-            }
-            CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER());
-//            long[] timeRange = getTimeRange(table);
-            Long countTotal = getCountTotal(table);
-            for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
-//                String sql = getQuerySql(timeRange, i, table);
-                String sql = getQuerySql(countTotal, i, table);
-                ReadHistoryArangoData<T> readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
-                threadPool.executor(readHistoryArangoData);
-            }
-            countDownLatch.await();
-            long last = System.currentTimeMillis();
-            LOG.warn("读取" + table + " arangoDB 共耗时：" + (last - start));
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
-
-    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";
-    }
-
-}

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

@@ -1,125 +0,0 @@
-package cn.ac.iie.service.read;
-
-import cn.ac.iie.config.ApplicationConfig;
-import cn.ac.iie.utils.ArangoDBConnect;
-import com.arangodb.ArangoCursor;
-import com.arangodb.entity.BaseDocument;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-
-
-/**
- * @author wlh
- * 多线程全量读取arangoDb历史数据，封装到map
- */
-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);
-    static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR();
-
-    public static final HashSet<String> PROTOCOL_SET;
-
-    static {
-        PROTOCOL_SET = new HashSet<>();
-        PROTOCOL_SET.add("HTTP");
-        PROTOCOL_SET.add("TLS");
-        PROTOCOL_SET.add("DNS");
-    }
-
-    private ArangoDBConnect arangoConnect;
-    private String query;
-    private ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map;
-    private Class<T> type;
-    private String table;
-    private CountDownLatch countDownLatch;
-
-    public ReadHistoryArangoData(ArangoDBConnect arangoConnect,
-                                 String query,
-                                 ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,
-                                 Class<T> type,
-                                 String table,
-                                 CountDownLatch countDownLatch) {
-        this.arangoConnect = arangoConnect;
-        this.query = query;
-        this.map = map;
-        this.type = type;
-        this.table = table;
-        this.countDownLatch = countDownLatch;
-    }
-
-    @Override
-    public void run() {
-        try {
-            long s = System.currentTimeMillis();
-            ArangoCursor<T> docs = arangoConnect.executorQuery(query, type);
-            if (docs != null) {
-                List<T> baseDocuments = docs.asListRemaining();
-                int i = 0;
-                for (T doc : baseDocuments) {
-                    String key = doc.getKey();
-                    switch (table) {
-                        case "R_LOCATE_FQDN2IP":
-                            updateProtocolDocument(doc);
-                            deleteDistinctClientIpByTime(doc);
-                            break;
-                        case "R_VISIT_IP2FQDN":
-                            updateProtocolDocument(doc);
-                            break;
-                        default:
-                    }
-                    int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER();
-                    ConcurrentHashMap<String, T> tmpMap = map.get(hashCode);
-                    tmpMap.put(key, doc);
-                    i++;
-                }
-                long l = System.currentTimeMillis();
-                LOG.warn(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
-            }
-        }catch (Exception e){
-            e.printStackTrace();
-        }finally {
-            countDownLatch.countDown();
-            LOG.warn("本线程读取完毕，剩余线程数量："+countDownLatch.getCount());
-        }
-    }
-
-    private void updateProtocolDocument(T doc) {
-        if (doc.getProperties().containsKey("PROTOCOL_TYPE")) {
-            for (String protocol : PROTOCOL_SET) {
-                String protocolRecent = protocol + "_CNT_RECENT";
-                ArrayList<Long> cntRecent = (ArrayList<Long>) doc.getAttribute(protocolRecent);
-                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;
-                doc.addAttribute(protocolRecent, cntRecentsDst);
-            }
-        }
-    }
-
-    private void deleteDistinctClientIpByTime(T doc) {
-        ArrayList<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
-        ArrayList<Long> distCipTs = (ArrayList<Long>) doc.getAttribute("DIST_CIP_TS");
-        distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
-        Collections.sort(distCipTs);
-        int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);
-        String[] distCipArr = new String[index];
-        long[] disCipTsArr = new long[index];
-        if (distCip.size() + 1 == distCipTs.size()){
-            for (int i = 0; i < index; i++) {
-                distCipArr[i] = distCip.get(i);
-                disCipTsArr[i] = distCipTs.get(i);
-            }
-        }
-        doc.updateAttribute("DIST_CIP", distCipArr);
-        doc.updateAttribute("DIST_CIP_TS", disCipTsArr);
-    }
-
-}

ip-learning-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java

@@ -22,11 +22,12 @@ public class ArangoDBConnect {
     private static final Logger LOG = LoggerFactory.getLogger(ArangoDBConnect.class);
     private static ArangoDB arangoDB = null;
     private static ArangoDBConnect conn = null;
+
     static {
         getArangoDatabase();
     }
 
-    private static void getArangoDatabase(){
+    private static void getArangoDatabase() {
         arangoDB = new ArangoDB.Builder()
                 .maxConnections(ApplicationConfig.THREAD_POOL_NUMBER())
                 .host(ApplicationConfig.ARANGODB_HOST(), ApplicationConfig.ARANGODB_PORT())
@@ -35,82 +36,80 @@ public class ArangoDBConnect {
                 .build();
     }
 
-    public static synchronized ArangoDBConnect getInstance(){
-        if (null == conn){
+    public static synchronized ArangoDBConnect getInstance() {
+        if (null == conn) {
             conn = new ArangoDBConnect();
         }
         return conn;
     }
 
-    private ArangoDatabase getDatabase(){
+    private ArangoDatabase getDatabase() {
         return arangoDB.db(ApplicationConfig.ARANGODB_DB_NAME());
     }
 
-    public void clean(){
+    public void clean() {
         try {
-            if (arangoDB != null){
+            if (arangoDB != null) {
                 arangoDB.shutdown();
             }
-        }catch (Exception e){
+        } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
-    public <T> ArangoCursor<T> executorQuery(String query,Class<T> type){
+    public <T> ArangoCursor<T> executorQuery(String query, Class<T> type) {
         ArangoDatabase database = getDatabase();
         Map<String, Object> bindVars = new MapBuilder().get();
         AqlQueryOptions options = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL());
         try {
             return database.query(query, bindVars, options, type);
-        }catch (Exception e){
+        } catch (Exception e) {
             e.printStackTrace();
             return null;
-        }finally {
+        } finally {
             bindVars.clear();
         }
     }
 
     @Deprecated
-    public <T> void insertAndUpdate(ArrayList<T> docInsert,ArrayList<T> docUpdate,String collectionName){
+    public <T> void insertAndUpdate(ArrayList<T> docInsert, ArrayList<T> docUpdate, String collectionName) {
         ArangoDatabase database = getDatabase();
         try {
             ArangoCollection collection = database.collection(collectionName);
-            if (!docInsert.isEmpty()){
+            if (!docInsert.isEmpty()) {
                 collection.importDocuments(docInsert);
             }
-            if (!docUpdate.isEmpty()){
+            if (!docUpdate.isEmpty()) {
                 collection.replaceDocuments(docUpdate);
             }
-        }catch (Exception e){
-            System.out.println("更新失败");
-            e.printStackTrace();
-        }finally {
+        } catch (Exception e) {
+            LOG.error("update failure：" + e.toString());
+        } finally {
             docInsert.clear();
             docInsert.clear();
         }
     }
 
-    public <T> void overwrite(ArrayList<T> docOverwrite,String collectionName){
+    public <T> void overwrite(ArrayList<T> docOverwrite, String collectionName) {
         ArangoDatabase database = getDatabase();
         try {
             ArangoCollection collection = database.collection(collectionName);
-            if (!docOverwrite.isEmpty()){
+            if (!docOverwrite.isEmpty()) {
                 DocumentCreateOptions documentCreateOptions = new DocumentCreateOptions();
                 documentCreateOptions.overwrite(true);
                 documentCreateOptions.silent(true);
                 MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
                 Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
-                for (ErrorEntity errorEntity:errors){
-                    LOG.warn("写入arangoDB异常："+errorEntity.getErrorMessage());
+                for (ErrorEntity errorEntity : errors) {
+                    LOG.warn("write arangoDB error：" + errorEntity.getErrorMessage());
                 }
             }
-        }catch (Exception e){
-            System.out.println("更新失败："+e.toString());
-        }finally {
+        } catch (Exception e) {
+            LOG.error("update failure：" + e.toString());
+        } finally {
             docOverwrite.clear();
         }
     }
 
 
-
 }

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

@@ -1,39 +1,45 @@
 #spark任务配置
 spark.sql.shuffle.partitions=10
 spark.executor.memory=4g
+spark.executor.cores=1
+spark.cores.max=10
+spark.local.dir=./tmp
 spark.app.name=test
 spark.network.timeout=300s
 spark.serializer=org.apache.spark.serializer.KryoSerializer
 master=local[*]
 #spark读取clickhouse配置
-spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.67: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=ceiec2019
+spark.read.clickhouse.password=galaxy2019
 spark.read.clickhouse.numPartitions=5
-spark.read.clickhouse.fetchsize=10000
+spark.read.clickhouse.fetchsize=10
 spark.read.clickhouse.partitionColumn=LAST_FOUND_TIME
+spark.read.clickhouse.session.table=session_record
+spark.read.clickhouse.radius.table=radius_record
 clickhouse.socket.timeout=300000
 #arangoDB配置
-arangoDB.host=192.168.40.182
+arangoDB.host=192.168.44.12
 arangoDB.port=8529
-arangoDB.user=upsert
-arangoDB.password=ceiec2018
-arangoDB.DB.name=ip-learning-test-0
-#arangoDB.DB.name=iplearn_media_domain
+arangoDB.user=root
+arangoDB.password=galaxy_2019
+arangoDB.DB.name=tsg_galaxy_v3
 arangoDB.ttl=3600
 
 thread.pool.number=10
 
 #读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
-clickhouse.time.limit.type=1
-read.clickhouse.max.time=1603785961
-read.clickhouse.min.time=1603354682
+clickhouse.time.limit.type=0
+read.clickhouse.max.time=1634902508
+read.clickhouse.min.time=1631759985
 
-arangoDB.read.limit=1
 update.arango.batch=10000
 
 distinct.client.ip.num=10000
 recent.count.hour=24
 
 update.interval=3600
+arangodb.total.num=20000000
+#读取radius时间范围,与radius任务执行周期一致,单位:分钟
+read.radius.granularity=-30

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

@@ -4,20 +4,15 @@ log4j.logger.org.apache.http.wire=OFF
 
 #Log4j
 log4j.rootLogger=info,console,file
-# <20><><EFBFBD><EFBFBD>̨<EFBFBD><CCA8>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
 log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.Threshold=warn
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
 log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] [Thread\:%t] %l %x - <%m>%n
 
-# <20>ļ<EFBFBD><C4BC><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
 log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
 log4j.appender.file.Threshold=warn
 log4j.appender.file.encoding=UTF-8
 log4j.appender.file.Append=true
-#·<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><D8B2><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-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala

@@ -7,11 +7,14 @@ object ApplicationConfig {
 
   val SPARK_SQL_SHUFFLE_PARTITIONS: Int = config.getInt("spark.sql.shuffle.partitions")
   val SPARK_EXECUTOR_MEMORY: String = config.getString("spark.executor.memory")
+  val SPARK_EXECUTOR_CORES: String = config.getString("spark.executor.cores")
+  val SPARK_CORES_MAX: String = config.getString("spark.cores.max")
   val SPARK_APP_NAME: String = config.getString("spark.app.name")
   val SPARK_NETWORK_TIMEOUT: String = config.getString("spark.network.timeout")
 //  val REPARTITION_NUMBER: Int = config.getInt("repartitionNumber")
   val MASTER: String = config.getString("master")
   val SPARK_SERIALIZER: String = config.getString("spark.serializer")
+  val SPARK_LOCAL_DIR: String = config.getString("spark.local.dir")
 
   val NUMPARTITIONS: String = config.getString("spark.read.clickhouse.numPartitions")
   val SPARK_READ_CLICKHOUSE_URL: String = config.getString("spark.read.clickhouse.url")
@@ -20,6 +23,8 @@ object ApplicationConfig {
   val SPARK_READ_CLICKHOUSE_PASSWORD: String = config.getString("spark.read.clickhouse.password")
   val SPARK_READ_CLICKHOUSE_FETCHSIZE: String = config.getString("spark.read.clickhouse.fetchsize")
   val SPARK_READ_CLICKHOUSE_PARTITIONCOLUMN: String = config.getString("spark.read.clickhouse.partitionColumn")
+  val SPARK_READ_CLICKHOUSE_SESSION_TABLE: String = config.getString("spark.read.clickhouse.session.table")
+  val SPARK_READ_CLICKHOUSE_RADIUS_TABLE: String = config.getString("spark.read.clickhouse.radius.table")
 
   val ARANGODB_HOST: String= config.getString("arangoDB.host")
   val ARANGODB_PORT: Int = config.getInt("arangoDB.port")
@@ -36,11 +41,14 @@ 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: Int = config.getInt("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")
 
   val UPDATE_INTERVAL: Int = config.getInt("update.interval")
 
+  val ARANGODB_TOTAL_NUM: Long = config.getLong("arangodb.total.num")
+
+  val READ_RADIUS_GRANULARITY: Int = config.getInt("read.radius.granularity")
+
 }

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

@@ -0,0 +1,22 @@
+package cn.ac.iie.dao
+
+import cn.ac.iie.config.ApplicationConfig
+import cn.ac.iie.spark.ArangoSpark
+import cn.ac.iie.spark.rdd.{ArangoRdd, ReadOptions}
+import cn.ac.iie.utils.SparkSessionUtil.sparkContext
+import org.slf4j.LoggerFactory
+
+import scala.reflect.ClassTag
+
+object BaseArangoData {
+  private val LOG = LoggerFactory.getLogger(BaseArangoData.getClass)
+  private val options = ReadOptions(ApplicationConfig.ARANGODB_DB_NAME)
+
+  def loadArangoRdd[T: ClassTag](name:String): ArangoRdd[T] ={
+    val value = ArangoSpark.load[T](sparkContext, name, options)
+
+    LOG.warn(s"read $name arangoDb: ${value.count()}")
+    value
+  }
+
+}

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

@@ -2,16 +2,19 @@ package cn.ac.iie.dao
 
 import cn.ac.iie.config.ApplicationConfig
 import cn.ac.iie.utils.SparkSessionUtil.spark
+import com.zdjizhi.utils.DateUtils
 import org.apache.spark.sql.DataFrame
 import org.slf4j.LoggerFactory
 
+import java.util.Date
+
 object BaseClickhouseData {
   private val LOG = LoggerFactory.getLogger(BaseClickhouseData.getClass)
 
   val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60
   private val timeLimit: (Long, Long) = getTimeLimit
 
-  private def initClickhouseData(sql:String): DataFrame  ={
+  private def initClickhouseData(sql: String): DataFrame = {
 
     val dataFrame: DataFrame = spark.read.format("jdbc")
       .option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL)
@@ -24,7 +27,7 @@ object BaseClickhouseData {
       .option("lowerBound", timeLimit._2)
       .option("upperBound", timeLimit._1)
       .option("fetchsize", ApplicationConfig.SPARK_READ_CLICKHOUSE_FETCHSIZE)
-      .option("socket_timeout",ApplicationConfig.CLICKHOUSE_SOCKET_TIMEOUT)
+      .option("socket_timeout", ApplicationConfig.CLICKHOUSE_SOCKET_TIMEOUT)
       .load()
     dataFrame.printSchema()
     dataFrame.createOrReplaceGlobalTempView("dbtable")
@@ -32,150 +35,23 @@ object BaseClickhouseData {
     dataFrame
   }
 
-  def loadConnectionDataFromCk(): Unit ={
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+  def getVertexFqdnDf: DataFrame = {
+    val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
     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
-        |FROM
-        | connection_record_log
-        |WHERE $where) as dbtable
-      """.stripMargin
-
-    LOG.warn(sql)
-    initClickhouseData(sql)
-  }
-
-  private def loadRadiusDataFromCk(): Unit ={
-    val where =
-      s"""
-        | common_recv_time >= ${timeLimit._2}
-        | AND common_recv_time < ${timeLimit._1}
-        | AND common_subscriber_id != ''
-        | AND radius_framed_ip != ''
-        | AND radius_packet_type = 4
-        | AND radius_acct_status_type = 1
-      """.stripMargin
-    val sql =
-      s"""
-        |(SELECT
-        | common_subscriber_id,radius_framed_ip,common_recv_time
-        |FROM
-        | tsg_galaxy_v3.radius_record_log
-        |WHERE
-        | $where) as dbtable
-      """.stripMargin
-    LOG.warn(sql)
-    initClickhouseData(sql)
-  }
-
-  /*
-  def getVertexIpDf: DataFrame ={
-    loadConnectionDataFromCk()
-    val sql =
-      """
-        |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 vertexIpDf = spark.sql(sql)
-    vertexIpDf.printSchema()
-    vertexIpDf
-  }
-
-  def getRelationFqdnLocateIpDf: DataFrame ={
-    loadConnectionDataFromCk()
-    val 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,
-        | collect_set(common_client_ip) AS DIST_CIP_RECENT,
-        | 'TLS' AS  schema_type
-        |FROM
-        | global_temp.dbtable
-        |WHERE
-        | common_schema_type = 'SSL'
-        |GROUP BY
-        | ssl_sni,common_server_ip
-      """.stripMargin
-
-    val 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,
-        | collect_set(common_client_ip) AS DIST_CIP_RECENT,
-        | 'HTTP' AS schema_type
-        |FROM
-        | global_temp.dbtable
-        |WHERE
-        | common_schema_type = 'HTTP'
-        |GROUP BY
-        | http_host,common_server_ip
-      """.stripMargin
-    val sql = s"SELECT * FROM (($sslSql) UNION ALL ($httpSql)) WHERE FQDN != ''"
-
-    LOG.warn(sql)
-    val relationFqdnLocateIpDf = spark.sql(sql)
-    relationFqdnLocateIpDf.printSchema()
-    relationFqdnLocateIpDf
-  }
-  */
-
-  def getVertexFqdnDf: DataFrame ={
-    val sql =
-      """
-        |(SELECT
-        | FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME
-        |FROM
-        | ((SELECT
-        |    ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
-        |   FROM tsg_galaxy_v3.connection_record_log
-        |   WHERE common_schema_type = 'SSL' GROUP BY ssl_sni
-        |  )UNION ALL
-        |  (SELECT
-        |    http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
-        |   FROM tsg_galaxy_v3.connection_record_log
-        |   WHERE common_schema_type = 'HTTP' GROUP BY http_host))
-        |GROUP BY FQDN HAVING FQDN != '') as dbtable
+         |(SELECT
+         | FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
+         |FROM
+         | ((SELECT
+         |    ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
+         |   FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
+         |   WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,vsys_id
+         |  )UNION ALL
+         |  (SELECT
+         |    http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME,vsys_id AS VSYS_ID
+         |   FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
+         |   WHERE $where and decoded_as = 'HTTP' GROUP BY http_host,vsys_id))
+         |GROUP BY FQDN,VSYS_ID HAVING FQDN != '') as dbtable
       """.stripMargin
     LOG.warn(sql)
     val frame = initClickhouseData(sql)
@@ -183,31 +59,33 @@ object BaseClickhouseData {
     frame
   }
 
-  def getVertexIpDf: DataFrame ={
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+  def getVertexIpDf: DataFrame = {
+    val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
     val sql =
       s"""
          |(SELECT * FROM
-         |((SELECT common_client_ip AS IP,MIN(common_recv_time) AS FIRST_FOUND_TIME,
-         |MAX(common_recv_time) AS LAST_FOUND_TIME,
+         |((SELECT client_ip AS IP,MIN(recv_time) AS FIRST_FOUND_TIME,
+         |MAX(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)[2] as common_link_info,
+         |SUM(sent_bytes+received_bytes) as BYTES_SUM,
+         |'' as common_link_info,
          |'client' as ip_type
-         |FROM tsg_galaxy_v3.connection_record_log
+         |,vsys_id AS VSYS_ID
+         |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
          |where $where
-         |group by common_client_ip)
+         |group by client_ip,vsys_id)
          |UNION ALL
-         |(SELECT common_server_ip AS IP,
-         |MIN(common_recv_time) AS FIRST_FOUND_TIME,
-         |MAX(common_recv_time) AS LAST_FOUND_TIME,
+         |(SELECT server_ip AS IP,
+         |MIN(recv_time) AS FIRST_FOUND_TIME,
+         |MAX(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)[2] as common_link_info,
+         |SUM(sent_bytes+received_bytes) as BYTES_SUM,
+         |'' as common_link_info,
          |'server' as ip_type
-         |FROM tsg_galaxy_v3.connection_record_log
+         |,vsys_id AS VSYS_ID
+         |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
          |where $where
-         |group by common_server_ip))) as dbtable
+         |group by server_ip,vsys_id))) as dbtable
       """.stripMargin
     LOG.warn(sql)
     val frame = initClickhouseData(sql)
@@ -216,20 +94,20 @@ object BaseClickhouseData {
   }
 
 
-  def getRelationFqdnLocateIpDf: DataFrame ={
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+  def getRelationFqdnLocateIpDf: DataFrame = {
+    val where = "recv_time >= " + timeLimit._2 + " AND 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)
+         |((SELECT ssl_sni AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
+         |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'TLS' AS decoded_as_list, vsys_id AS VSYS_ID
+         |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
+         |WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,server_ip,vsys_id)
          |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))
+         |(SELECT http_host AS FQDN,server_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
+         |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'HTTP' AS decoded_as_list,vsys_id AS VSYS_ID
+         |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
+         |WHERE $where and decoded_as = 'HTTP' GROUP BY http_host,server_ip,vsys_id))
          |WHERE FQDN != '') as dbtable
       """.stripMargin
     LOG.warn(sql)
@@ -238,20 +116,22 @@ object BaseClickhouseData {
     frame
   }
 
-  def getRelationSubidLocateIpDf: DataFrame ={
+  def getRelationSubidLocateIpDf: DataFrame = {
     val where =
       s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
+         | common_recv_time >= ${getRadiusTimeRange._2}
+         | AND common_recv_time < ${getRadiusTimeRange._1}
          | AND common_subscriber_id != ''
          | AND radius_framed_ip != ''
+         | AND radius_packet_type = 4
+         | AND radius_acct_status_type = 1
       """.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
+         |SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
+         |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_RADIUS_TABLE}
+         |WHERE $where GROUP BY common_subscriber_id,radius_framed_ip,common_vsys_id
          |) as dbtable
       """.stripMargin
     LOG.warn(sql)
@@ -260,19 +140,22 @@ object BaseClickhouseData {
     frame
   }
 
-  def getVertexSubidDf: DataFrame ={
+  def getVertexSubidDf: DataFrame = {
     val where =
       s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
+         | common_recv_time >= ${getRadiusTimeRange._2}
+         | AND common_recv_time < ${getRadiusTimeRange._1}
          | AND common_subscriber_id != ''
          | AND radius_framed_ip != ''
+         | AND radius_packet_type = 4
+         | AND radius_acct_status_type = 1
       """.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
+         |SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_RADIUS_TABLE}
+         |WHERE $where GROUP BY common_subscriber_id,common_vsys_id
          |)as dbtable
       """.stripMargin
     LOG.warn(sql)
@@ -281,18 +164,21 @@ object BaseClickhouseData {
     frame
   }
 
-  def getVertexFramedIpDf: DataFrame ={
+  def getVertexFramedIpDf: DataFrame = {
     val where =
       s"""
-         | common_recv_time >= ${timeLimit._2}
-         | AND common_recv_time < ${timeLimit._1}
+         | common_recv_time >= ${getRadiusTimeRange._2}
+         | AND common_recv_time < ${getRadiusTimeRange._1}
          | AND common_subscriber_id != ''
          | AND radius_framed_ip != ''
+         | AND radius_packet_type = 4
+         | AND radius_acct_status_type = 1
       """.stripMargin
     val sql =
       s"""
          |(
-         |SELECT DISTINCT radius_framed_ip,common_recv_time as LAST_FOUND_TIME FROM radius_record_log WHERE $where
+         |SELECT radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,common_vsys_id AS VSYS_ID FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_RADIUS_TABLE} WHERE $where
+         |GROUP BY radius_framed_ip,common_vsys_id
          |)as dbtable
       """.stripMargin
     LOG.warn(sql)
@@ -301,8 +187,19 @@ object BaseClickhouseData {
     frame
   }
 
+  private def getRadiusTimeRange: (Long, Long) = {
+    val date = DateUtils.getTimeFloor(new Date(System.currentTimeMillis()), "PT1M")
+    val max = date.getTime / 1000
+    val min = DateUtils.getSomeMinute(date, ApplicationConfig.READ_RADIUS_GRANULARITY).getTime / 1000
+    (max, min)
+  }
 
-  private def getTimeLimit: (Long,Long) ={
+  def main(args: Array[String]): Unit = {
+    println(getRadiusTimeRange)
+    println(getRadiusTimeRange._2 - getRadiusTimeRange._1)
+  }
+
+  private def getTimeLimit: (Long, Long) = {
     var maxTime = 0L
     var minTime = 0L
     ApplicationConfig.CLICKHOUSE_TIME_LIMIT_TYPE match {

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

@@ -5,6 +5,6 @@ import cn.ac.iie.service.update.UpdateDocument
 object IpLearningApplication {
 
   def main(args: Array[String]): Unit = {
-    UpdateDocument.update()
+    UpdateDocument.ipLearning()
   }
 }

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

@@ -0,0 +1,12 @@
+package cn.ac.iie.main
+
+import cn.ac.iie.service.update.UpdateDocument
+
+@Deprecated
+object IpRecommendApplication {
+
+  def main(args: Array[String]): Unit = {
+    UpdateDocument.ipRecommend()
+  }
+
+}

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

@@ -0,0 +1,12 @@
+package cn.ac.iie.main
+
+import cn.ac.iie.service.update.UpdateDocument
+
+@Deprecated
+object SubscriberRecommendApplication {
+
+  def main(args: Array[String]): Unit = {
+    UpdateDocument.subscriberRecommend()
+  }
+
+}

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

@@ -1,41 +1,35 @@
 package cn.ac.iie.service.transform
 
-import java.util.regex.Pattern
-
 import cn.ac.iie.config.ApplicationConfig
-import cn.ac.iie.dao.BaseClickhouseData
-import cn.ac.iie.spark.ArangoSpark
+import cn.ac.iie.dao.{BaseArangoData, BaseClickhouseData}
 import cn.ac.iie.spark.partition.CustomPartitioner
-import cn.ac.iie.spark.rdd.ReadOptions
+import cn.ac.iie.spark.rdd.ArangoRdd
 import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.functions._
 import org.slf4j.LoggerFactory
-import cn.ac.iie.utils.SparkSessionUtil._
+
+import java.util.regex.Pattern
 
 object MergeDataFrame {
   private val LOG = LoggerFactory.getLogger(MergeDataFrame.getClass)
   private val pattern = Pattern.compile("^[\\d]*$")
-  private val options = ReadOptions(ApplicationConfig.ARANGODB_DB_NAME)
 
-  def mergeVertexFqdn(): RDD[(String, (Option[BaseDocument], Option[Row]))] ={
-    val fqdnAccmu = getLongAccumulator("FQDN Accumulator")
-    val fqdnRddRow = BaseClickhouseData.getVertexFqdnDf
-      .rdd.filter(row => isDomain(row.getAs[String](0))).map(row => {
-      fqdnAccmu.add(1)
+  def mergeVertexFqdn(): RDD[(String, (Option[BaseDocument], Row))] = {
+    val fqdnRddRow: RDD[(String, Row)] = BaseClickhouseData.getVertexFqdnDf
+      .repartition().rdd.filter(row => isDomain(row.getAs[String](0))).map(row => {
       (row.getAs[String]("FQDN"), row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
-    fqdnRddRow.cache()
-    val fqdnRddDoc = ArangoSpark.load[BaseDocument](sparkContext,"FQDN",options)
+    }) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
 
-    fqdnRddDoc.map(doc => (doc.getKey, doc)).fullOuterJoin(fqdnRddRow)
+    val fqdnRddDoc: ArangoRdd[BaseDocument] = BaseArangoData.loadArangoRdd[BaseDocument]("FQDN")
+
+    fqdnRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnRddRow)
   }
 
-  def mergeVertexIp(): RDD[(String, (Option[BaseDocument], Option[Row]))]={
-    val ipAccum = getLongAccumulator("IP Accumulator")
+  def mergeVertexIp(): RDD[(String, (Option[BaseDocument], Row))] = {
     val vertexIpDf = BaseClickhouseData.getVertexIpDf
-    val frame = vertexIpDf.groupBy("IP").agg(
+    val frame = vertexIpDf.repartition().groupBy("IP", "VSYS_ID").agg(
       min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
       max("LAST_FOUND_TIME").alias("LAST_FOUND_TIME"),
       collect_list("SESSION_COUNT").alias("SESSION_COUNT_LIST"),
@@ -43,79 +37,83 @@ object MergeDataFrame {
       collect_list("ip_type").alias("ip_type_list"),
       last("common_link_info").alias("common_link_info")
     )
-    val ipRddRow = frame.rdd.map(row => {
-      ipAccum.add(1)
-      (row.getAs[String]("IP"), row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
-    val ipRddDoc = ArangoSpark.load[BaseDocument](sparkContext,"IP",options)
 
-    ipRddDoc.map(doc => (doc.getKey, doc)).fullOuterJoin(ipRddRow)
+    val ipRddRow = frame.rdd.map(row => {
+      val vsysId = row.getAs[Integer]("VSYS_ID").toLong
+      val ip = row.getAs[String]("IP")
+      ( ip + "-" + vsysId, row)
+    }) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
+    val ipRddDoc = BaseArangoData.loadArangoRdd[BaseDocument]("IP")
+    ipRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(ipRddRow)
 
   }
 
-  def mergeRelationFqdnLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Option[Row]))] ={
-    val fqdnLocIpAccum = getLongAccumulator("R_LOCATE_FQDN2IP Accumulator")
-    val frame = BaseClickhouseData.getRelationFqdnLocateIpDf.filter(row => isDomain(row.getAs[String]("FQDN")))
-      .groupBy("FQDN", "common_server_ip")
+  def mergeRelationFqdnLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
+    val frame = BaseClickhouseData.getRelationFqdnLocateIpDf
+      .repartition()
+      .filter(row => isDomain(row.getAs[String]("FQDN")))
+      .groupBy("FQDN", "server_ip", "VSYS_ID")
       .agg(
         min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
         max("LAST_FOUND_TIME").alias("LAST_FOUND_TIME"),
         collect_list("COUNT_TOTAL").alias("COUNT_TOTAL_LIST"),
-        collect_list("schema_type").alias("schema_type_list"),
+        collect_list("decoded_as_list").alias("decoded_as_list"),
         collect_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT")
       )
-    val fqdnLocIpRddRow = frame.rdd.map(row => {
-      val fqdn = row.getAs[String]("FQDN")
-      val serverIp = row.getAs[String]("common_server_ip")
-      val key = fqdn.concat("-" + serverIp)
-      fqdnLocIpAccum.add(1)
-      (key, row)
-    }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
-    val fqdnLocIpRddDoc = ArangoSpark.load[BaseEdgeDocument](sparkContext,"R_LOCATE_FQDN2IP",options)
 
-    fqdnLocIpRddDoc.map(doc => (doc.getKey, doc)).fullOuterJoin(fqdnLocIpRddRow)
+    val fqdnLocIpRddRow = frame.rdd
+      .filter(row => {
+        // 检查 server_ip 和 VSYS_ID 是否为 null
+        val serverIp = row.getAs[String]("server_ip")
+        val vsysId = row.getAs[Integer]("VSYS_ID")
+        serverIp != null && vsysId != null
+      })
+      .map(row => {
+        val fqdn = row.getAs[String]("FQDN")
+        val serverIp = row.getAs[String]("server_ip")
+        val vsysId = row.getAs[Integer]("VSYS_ID").toLong
+        val key = fqdn.concat("-" + serverIp + "-" + vsysId)
+        (key, row)
+      })
+    val fqdnLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_FQDN2IP")
+    fqdnLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnLocIpRddRow)
 
   }
 
-  def mergeRelationSubidLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Option[Row]))] ={
-    val subidLocIpAccum = getLongAccumulator("R_LOCATE_SUBSCRIBER2IP Accumulator")
+  def mergeRelationSubidLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
     val subidLocIpRddRow = 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)
-      subidLocIpAccum.add(1)
+      val vsysId = row.getAs[Long]("VSYS_ID")
+      val key = commonSubscriberId.concat("-" + ip + "-" + vsysId)
       (key, row)
     }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
-    val subidLocIpRddDoc = ArangoSpark.load[BaseEdgeDocument](sparkContext,"R_LOCATE_SUBSCRIBER2IP",options)
+    val subidLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_SUBSCRIBER2IP")
 
-    subidLocIpRddDoc.map(doc => (doc.getKey, doc)).fullOuterJoin(subidLocIpRddRow)
+    subidLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(subidLocIpRddRow)
   }
 
-  def mergeVertexSubid(): RDD[(String, (Option[BaseDocument], Option[Row]))] ={
-    val subidAccum = getLongAccumulator("SUBSCRIBER Accumulator")
+  def mergeVertexSubid(): RDD[(String, (Option[BaseDocument], Row))] = {
     val subidRddRow = BaseClickhouseData.getVertexSubidDf
       .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
       .rdd.map(row => {
       val commonSubscriberId = row.getAs[String]("common_subscriber_id")
-      subidAccum.add(1)
       (commonSubscriberId, row)
     }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
 
-    val subidRddDoc = ArangoSpark.load[BaseDocument](sparkContext,"SUBSCRIBER",options)
+    val subidRddDoc = BaseArangoData.loadArangoRdd[BaseDocument]("SUBSCRIBER")
 
-    subidRddDoc.map(doc => (doc.getKey, doc)).fullOuterJoin(subidRddRow)
+    subidRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(subidRddRow)
 
   }
 
-  def mergeVertexFrameIp: RDD[Row] ={
-    val framedIpAccum = getLongAccumulator("framed ip Accumulator")
+  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")
-      framedIpAccum.add(1)
       (ip, row)
     }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
     values
@@ -129,7 +127,7 @@ object MergeDataFrame {
 
       val fqdnArr = fqdn.split(":")(0).split("\\.")
 
-      if (fqdnArr.length != 4){
+      if (fqdnArr.length != 4) {
         return true
       }
       for (f <- fqdnArr) {
@@ -144,7 +142,7 @@ object MergeDataFrame {
       }
     } catch {
       case e: Exception =>
-        LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString)
+        LOG.error("Domain name resolution " + fqdn + " failure：\n" + e.toString)
     }
     false
 

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

@@ -2,47 +2,49 @@ package cn.ac.iie.service.update
 
 
 import java.util
-import scala.collection.JavaConversions._
 
+import scala.collection.JavaConversions._
 import cn.ac.iie.config.ApplicationConfig
-import cn.ac.iie.service.read.ReadHistoryArangoData
+import cn.ac.iie.dao.BaseClickhouseData
 import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
 
 import scala.collection.mutable
 import scala.collection.mutable.WrappedArray.ofRef
 
 object UpdateDocHandler {
-  val PROTOCOL_SET: Set[String] = Set("HTTP","TLS","DNS")
+  val PROTOCOL_SET: Set[String] = Set("HTTP", "TLS", "DNS")
 
-  def updateMaxAttribute(hisDoc: BaseDocument,newAttribute:Long,attributeName:String): Unit ={
-    if(hisDoc.getProperties.containsKey(attributeName)){
+  def updateMaxAttribute(hisDoc: BaseDocument, newAttribute: Long, attributeName: String): Unit = {
+    if (hisDoc.getProperties.containsKey(attributeName)) {
       var hisAttritube = hisDoc.getAttribute(attributeName).toString.toLong
-      if (newAttribute > hisAttritube){
+      if (newAttribute > hisAttritube) {
         hisAttritube = newAttribute
       }
-      hisDoc.addAttribute(attributeName,hisAttritube)
+      hisDoc.addAttribute(attributeName, hisAttritube)
     }
   }
 
-  def updateSumAttribute(hisDoc: BaseDocument,newAttribute:Long,attributeName:String): Unit ={
-    if (hisDoc.getProperties.containsKey(attributeName)){
+  def updateSumAttribute(hisDoc: BaseDocument, newAttribute: Long, attributeName: String): Unit = {
+    if (hisDoc.getProperties.containsKey(attributeName)) {
       val hisAttritube = hisDoc.getAttribute(attributeName).toString.toLong
-      hisDoc.addAttribute(attributeName,newAttribute+hisAttritube)
+      hisDoc.addAttribute(attributeName, newAttribute + hisAttritube)
     }
+
   }
 
-  def replaceAttribute(hisDoc: BaseDocument,newAttribute:String,attributeName:String): Unit ={
-    hisDoc.addAttribute(attributeName,newAttribute)
+  def replaceAttribute(hisDoc: BaseDocument, newAttribute: String, attributeName: String): Unit = {
+    //      hisDoc.addAttribute(attributeName,newAttribute)
+    hisDoc.updateAttribute(attributeName, newAttribute)
   }
 
-  def separateAttributeByIpType(ipTypeList:ofRef[String],
-                                sessionCountList:ofRef[AnyRef],
-                                bytesSumList:ofRef[AnyRef]): (Long,Long,Long,Long) ={
+  def separateAttributeByIpType(ipTypeList: ofRef[String],
+                                sessionCountList: ofRef[AnyRef],
+                                bytesSumList: ofRef[AnyRef]): (Long, Long, Long, Long) = {
     var serverSessionCount = 0L
     var serverBytesSum = 0L
     var clientSessionCount = 0L
     var clientBytesSum = 0L
-    if (ipTypeList.length == sessionCountList.length && ipTypeList.length == bytesSumList.length){
+    if (ipTypeList.length == sessionCountList.length && ipTypeList.length == bytesSumList.length) {
       sessionCountList.zip(bytesSumList).zip(ipTypeList).foreach(t => {
         t._2 match {
           case "server" =>
@@ -57,51 +59,51 @@ object UpdateDocHandler {
     (serverSessionCount, serverBytesSum, clientSessionCount, clientBytesSum)
   }
 
-  def separateAttributeByProtocol(schemaTypeList:ofRef[AnyRef],countTotalList:ofRef[AnyRef]): Map[String, Long] ={
+  def separateAttributeByProtocol(schemaTypeList: ofRef[AnyRef], countTotalList: ofRef[AnyRef]): Map[String, Long] = {
     var protocolMap: Map[String, Long] = Map()
-    if (schemaTypeList.length == countTotalList.length){
-      protocolMap = schemaTypeList.zip(countTotalList).map(t => (t._1.toString,t._2.toString.toLong)).toMap
+    if (schemaTypeList.length == countTotalList.length) {
+      protocolMap = schemaTypeList.zip(countTotalList).map(t => (t._1.toString, t._2.toString.toLong)).toMap
     }
     PROTOCOL_SET.foreach(protocol => {
-      if (!protocolMap.contains(protocol)){
+      if (!protocolMap.contains(protocol)) {
         protocolMap += (protocol -> 0L)
       }
     })
     protocolMap
   }
 
-  def updateProtocolAttritube(hisDoc:BaseEdgeDocument, protocolMap: Map[String, Long]): Unit ={
-    if (hisDoc.getProperties.containsKey("PROTOCOL_TYPE")){
+  def updateProtocolAttritube(hisDoc: BaseEdgeDocument, protocolMap: Map[String, Long]): Unit = {
+    if (hisDoc.getProperties.containsKey("PROTOCOL_TYPE")) {
       var protocolType = hisDoc.getAttribute("PROTOCOL_TYPE").toString
       protocolMap.foreach((t: (String, Long)) => {
-        if (t._2 > 0 && !protocolType.contains(t._1)){
-          protocolType = protocolType.concat(","+ t._1)
+        if (t._2 > 0 && !protocolType.contains(t._1)) {
+          protocolType = protocolType.concat("," + t._1)
         }
         val cntTotalName = t._1.concat("_CNT_TOTAL")
         val cntRecentName = t._1.concat("_CNT_RECENT")
         val cntRecent = hisDoc.getAttribute(cntRecentName).asInstanceOf[Array[Long]]
-        cntRecent.update(0,t._2)
-        updateSumAttribute(hisDoc,t._2,cntTotalName)
-        hisDoc.addAttribute(cntRecentName,cntRecent)
+        cntRecent.update(0, t._2)
+        updateSumAttribute(hisDoc, t._2, cntTotalName)
+        hisDoc.addAttribute(cntRecentName, cntRecent)
       })
-      hisDoc.addAttribute("PROTOCOL_TYPE",protocolType)
+      hisDoc.addAttribute("PROTOCOL_TYPE", protocolType)
     }
   }
 
-  def putProtocolAttritube(doc:BaseEdgeDocument, protocolMap: Map[String, Long]): Unit ={
+  def putProtocolAttritube(doc: BaseEdgeDocument, protocolMap: Map[String, Long]): Unit = {
     val protocolTypeBuilder = new mutable.StringBuilder()
     protocolMap.foreach(t => {
-      if (t._2 > 0){
-        protocolTypeBuilder.append(","+t._1)
+      if (t._2 > 0) {
+        protocolTypeBuilder.append("," + t._1)
       }
       val cntTotalName = t._1.concat("_CNT_TOTAL")
       val cntRecentName = t._1.concat("_CNT_RECENT")
       val cntRecent: Array[Long] = new Array[Long](ApplicationConfig.RECENT_COUNT_HOUR)
-      cntRecent.update(0,t._2)
-      doc.addAttribute(cntTotalName,t._2)
-      doc.addAttribute(cntRecentName,cntRecent)
+      cntRecent.update(0, t._2)
+      doc.addAttribute(cntTotalName, t._2)
+      doc.addAttribute(cntRecentName, cntRecent)
     })
-    doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",",""))
+    doc.addAttribute("PROTOCOL_TYPE", protocolTypeBuilder.toString().replaceFirst(",", ""))
   }
 
   def updateProtocolDocument(doc: BaseEdgeDocument): Unit = {
@@ -118,37 +120,37 @@ object UpdateDocHandler {
     }
   }
 
-  def mergeDistinctIp(distCipRecent:ofRef[String]): Array[String] ={
+  def mergeDistinctIp(distCipRecent: ofRef[String]): Array[String] = {
     distCipRecent.flatMap(str => {
-      str.replaceAll("\\[","")
-        .replaceAll("\\]","")
-        .replaceAll("\\'","")
+      str.replaceAll("\\[", "")
+        .replaceAll("\\]", "")
+        .replaceAll("\\'", "")
         .split(",")
     }).distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
   }
 
 
-  def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={
+  def putDistinctIp(doc: BaseEdgeDocument, newDistinctIp: Array[String]): Unit = {
     val map = newDistinctIp.map(ip => {
-      (ip, ReadHistoryArangoData.currentHour)
+      (ip, BaseClickhouseData.currentHour)
     }).toMap
-    doc.addAttribute("DIST_CIP",map.keys.toArray)
-    doc.addAttribute("DIST_CIP_TS",map.values.toArray)
+    doc.addAttribute("DIST_CIP", map.keys.toArray)
+    doc.addAttribute("DIST_CIP_TS", map.values.toArray)
   }
 
-  def updateDistinctIp(hisDoc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={
-    if (hisDoc.getProperties.containsKey("DIST_CIP") && hisDoc.getProperties.containsKey("DIST_CIP_TS")){
+  def updateDistinctIp(hisDoc: BaseEdgeDocument, newDistinctIp: Array[String]): Unit = {
+    if (hisDoc.getProperties.containsKey("DIST_CIP") && hisDoc.getProperties.containsKey("DIST_CIP_TS")) {
       val hisDistCip = hisDoc.getAttribute("DIST_CIP").asInstanceOf[util.ArrayList[String]]
       val hisDistCipTs = hisDoc.getAttribute("DIST_CIP_TS").asInstanceOf[util.ArrayList[Long]]
-      if (hisDistCip.length == hisDistCipTs.length){
+      if (hisDistCip.length == hisDistCipTs.length) {
         val distCipToTsMap: Map[String, Long] = hisDistCip.zip(hisDistCipTs).toMap
-        val muDistCipToTsMap: mutable.Map[String, Long] = mutable.Map(distCipToTsMap.toSeq:_*)
+        val muDistCipToTsMap: mutable.Map[String, Long] = mutable.Map(distCipToTsMap.toSeq: _*)
         newDistinctIp.foreach(cip => {
-          muDistCipToTsMap.put(cip,ReadHistoryArangoData.currentHour)
+          muDistCipToTsMap.put(cip, BaseClickhouseData.currentHour)
         })
         val resultMap = muDistCipToTsMap.toList.sortBy(-_._2).take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toMap
-        hisDoc.addAttribute("DIST_CIP",resultMap.keys.toArray)
-        hisDoc.addAttribute("DIST_CIP_TS",resultMap.values.toArray)
+        hisDoc.addAttribute("DIST_CIP", resultMap.keys.toArray)
+        hisDoc.addAttribute("DIST_CIP_TS", resultMap.values.toArray)
       }
     }
   }

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

@@ -17,20 +17,35 @@ object UpdateDocument {
   private val arangoManger: ArangoDBConnect = ArangoDBConnect.getInstance()
   private val LOG = LoggerFactory.getLogger(UpdateDocument.getClass)
 
-  def update(): Unit = {
+  def ipLearning(): Unit = {
     try {
-      updateDocument("FQDN", getVertexFqdnRow, mergeVertexFqdn)
-
-      updateDocument("SUBSCRIBER",getVertexSubidRow,mergeVertexSubid)
-
-      insertFrameIp()
-
-      updateDocument("R_LOCATE_SUBSCRIBER2IP",getRelationSubidLocateIpRow,mergeRelationSubidLocateIp)
-
       updateDocument("R_LOCATE_FQDN2IP", getRelationFqdnLocateIpRow, mergeRelationFqdnLocateIp)
+    } catch {
+      case e: Exception => e.printStackTrace()
+    } finally {
+      arangoManger.clean()
+      SparkSessionUtil.closeSpark()
+      System.exit(0)
+    }
+  }
 
+  def subscriberRecommend(): Unit = {
+    try {
+      updateDocument("SUBSCRIBER", getVertexSubidRow, mergeVertexSubid)
+      insertFrameIp()
+      updateDocument("R_LOCATE_SUBSCRIBER2IP", getRelationSubidLocateIpRow, mergeRelationSubidLocateIp)
+    } catch {
+      case e: Exception => e.printStackTrace()
+    } finally {
+      arangoManger.clean()
+      SparkSessionUtil.closeSpark()
+      System.exit(0)
+    }
+  }
+
+  def ipRecommend(): Unit = {
+    try {
       updateDocument("IP", getVertexIpRow, mergeVertexIp)
-
     } catch {
       case e: Exception => e.printStackTrace()
     } finally {
@@ -41,40 +56,48 @@ object UpdateDocument {
   }
 
   private def updateDocument[T <: BaseDocument](collName: String,
-                                                getDocumentRow: ((String, (Option[T], Option[Row]))) => T,
-                                                getJoinRdd: () => RDD[(String, (Option[T], Option[Row]))]
+                                                getDocumentRow: ((String, (Option[T], Row))) => T,
+                                                getJoinRdd: () => RDD[(String, (Option[T], Row))]
                                                ): Unit = {
     try {
       val start = System.currentTimeMillis()
       val joinRdd = getJoinRdd()
+
+      val fqdnAccmu = SparkSessionUtil.getLongAccumulator(s"$collName Accumulator")
+
       joinRdd.foreachPartition(iter => {
         val resultDocumentList = new util.ArrayList[T]
         var i = 0
         iter.foreach(row => {
-          val document = getDocumentRow(row)
-          if (document != null){
+          val document: T = getDocumentRow(row)
+          if (document != null) {
+            fqdnAccmu.add(1)
+//            println(document)
             resultDocumentList.add(document)
           }
           i += 1
           if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
             arangoManger.overwrite(resultDocumentList, collName)
-            LOG.warn(s"更新:$collName" + i)
+            LOG.warn(s"update $collName: " + i)
             i = 0
           }
         })
         if (i != 0) {
           arangoManger.overwrite(resultDocumentList, collName)
-          LOG.warn(s"更新$collName:" + i)
+          LOG.warn(s"update $collName: " + i)
         }
       })
+
+      LOG.warn(s"update $collName rows: ${fqdnAccmu.value}")
+
       val last = System.currentTimeMillis()
-      LOG.warn(s"更新$collName 时间：${last - start}")
+      LOG.warn(s"update $collName time: ${last - start}")
     } catch {
       case e: Exception => e.printStackTrace()
     }
   }
 
-  private def insertFrameIp(): Unit ={
+  private def insertFrameIp(): Unit = {
     mergeVertexFrameIp.foreachPartition(iter => {
       val resultDocumentList = new util.ArrayList[BaseDocument]
       var i = 0
@@ -84,142 +107,99 @@ object UpdateDocument {
         i += 1
         if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
           arangoManger.overwrite(resultDocumentList, "IP")
-          LOG.warn(s"更新:IP" + i)
+          LOG.warn(s"update IP: " + i)
           i = 0
         }
       })
       if (i != 0) {
         arangoManger.overwrite(resultDocumentList, "IP")
-        LOG.warn(s"更新IP:" + i)
+        LOG.warn(s"update IP: " + i)
       }
     })
   }
 
-  private def getVertexFrameipRow(row: Row): BaseDocument ={
+  private def getVertexFrameipRow(row: Row): BaseDocument = {
     val ip = row.getAs[String]("radius_framed_ip")
+    val vsysId = row.getAs[Long]("VSYS_ID")
     val document = new BaseDocument()
-    document.setKey(ip)
-    document.addAttribute("IP",ip)
+    document.setKey(ip + "-" + vsysId)
+    document.addAttribute("IP", ip)
     document
   }
 
-  private def getRelationSubidLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Option[Row]))): BaseEdgeDocument ={
+  private def getRelationSubidLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Row))): BaseEdgeDocument = {
 
     val subidLocIpDocOpt = joinRow._2._1
     var subidLocIpDoc = subidLocIpDocOpt match {
       case Some(doc) => doc
       case None => null
     }
-
-    val subidLocIpRowOpt = joinRow._2._2
-
-    val subidLocIpRow = subidLocIpRowOpt match {
-      case Some(r) => r
-      case None => null
-    }
-
-    if (subidLocIpRow != null){
+    val subidLocIpRow = joinRow._2._2
+    if (subidLocIpRow != null) {
       val subId = subidLocIpRow.getAs[String]("common_subscriber_id")
       val ip = subidLocIpRow.getAs[String]("radius_framed_ip")
       val lastFoundTime = subidLocIpRow.getAs[Long]("LAST_FOUND_TIME")
       val firstFoundTime = subidLocIpRow.getAs[Long]("FIRST_FOUND_TIME")
+      val vsysId = subidLocIpRow.getAs[Long]("VSYS_ID")
 
-      val key = subId.concat("-"+ip)
-      if (subidLocIpDoc != null){
-        updateMaxAttribute(subidLocIpDoc,lastFoundTime,"LAST_FOUND_TIME")
+      val key = subId.concat("-" + ip + "-" + vsysId)
+      if (subidLocIpDoc != null) {
+        subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
+        subidLocIpDoc.addAttribute("SUBSCRIBER", subId)
+        subidLocIpDoc.addAttribute("IP", ip)
+        updateMaxAttribute(subidLocIpDoc, lastFoundTime, "LAST_FOUND_TIME")
       } else {
         subidLocIpDoc = new BaseEdgeDocument()
         subidLocIpDoc.setKey(key)
-        subidLocIpDoc.setFrom("SUBSCRIBER/" + subId)
-        subidLocIpDoc.setTo("IP/" + ip)
-        subidLocIpDoc.addAttribute("SUBSCRIBER",subId)
-        subidLocIpDoc.addAttribute("IP",ip)
-        subidLocIpDoc.addAttribute("FIRST_FOUND_TIME",firstFoundTime)
-        subidLocIpDoc.addAttribute("LAST_FOUND_TIME",lastFoundTime)
+        subidLocIpDoc.setFrom("SUBSCRIBER/" + subId + "-" + vsysId)
+        subidLocIpDoc.setTo("IP/" + ip+ "-" + vsysId)
+        subidLocIpDoc.addAttribute("SUBSCRIBER", subId)
+        subidLocIpDoc.addAttribute("IP", ip)
+        subidLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
+        subidLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
+        subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
       }
     }
     subidLocIpDoc
   }
 
-  private def getVertexSubidRow(joinRow: (String, (Option[BaseDocument], Option[Row]))): BaseDocument ={
+  private def getVertexSubidRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
     val subidDocOpt = joinRow._2._1
     var subidDoc = subidDocOpt match {
       case Some(doc) => doc
       case None => null
     }
-
-    val subidRowOpt = joinRow._2._2
-
-    val subidRow = subidRowOpt match {
-      case Some(r) => r
-      case None => null
-    }
-
-    if (subidRow != null){
+    val subidRow = joinRow._2._2
+    if (subidRow != null) {
       val subId = subidRow.getAs[String]("common_subscriber_id")
       val subLastFoundTime = subidRow.getAs[Long]("LAST_FOUND_TIME")
       val subFirstFoundTime = subidRow.getAs[Long]("FIRST_FOUND_TIME")
-      if (subidDoc != null){
-        updateMaxAttribute(subidDoc,subLastFoundTime,"LAST_FOUND_TIME")
+      val vsysId = subidRow.getAs[Long]("VSYS_ID")
+      val key = subId.concat("-" + vsysId)
+
+      if (subidDoc != null) {
+        updateMaxAttribute(subidDoc, subLastFoundTime, "LAST_FOUND_TIME")
+        subidDoc.addAttribute("VSYS_ID", vsysId)
       } else {
         subidDoc = new BaseDocument()
-        subidDoc.setKey(subId)
-        subidDoc.addAttribute("SUBSCRIBER",subId)
-        subidDoc.addAttribute("FIRST_FOUND_TIME",subFirstFoundTime)
-        subidDoc.addAttribute("LAST_FOUND_TIME",subLastFoundTime)
+        subidDoc.setKey(key)
+        subidDoc.addAttribute("SUBSCRIBER", subId)
+        subidDoc.addAttribute("FIRST_FOUND_TIME", subFirstFoundTime)
+        subidDoc.addAttribute("LAST_FOUND_TIME", subLastFoundTime)
+        subidDoc.addAttribute("VSYS_ID", vsysId)
       }
     }
-
     subidDoc
   }
 
-  private def getVertexFqdnRow(joinRow: (String, (Option[BaseDocument], Option[Row]))): BaseDocument = {
-    val fqdnDocOpt = joinRow._2._1
-    var fqdnDoc = fqdnDocOpt match {
-      case Some(doc) => doc
-      case None => null
-    }
-
-    val fqdnRowOpt = joinRow._2._2
-
-    val fqdnRow = fqdnRowOpt match {
-      case Some(r) => r
-      case None => null
-    }
-
-    if (fqdnRow != null){
-      val fqdn = fqdnRow.getAs[String]("FQDN")
-      val lastFoundTime = fqdnRow.getAs[Long]("LAST_FOUND_TIME")
-      val firstFoundTime = fqdnRow.getAs[Long]("FIRST_FOUND_TIME")
-      if (fqdnDoc != null) {
-        updateMaxAttribute(fqdnDoc, lastFoundTime, "LAST_FOUND_TIME")
-      } else {
-        fqdnDoc = new BaseDocument
-        fqdnDoc.setKey(fqdn)
-        fqdnDoc.addAttribute("FQDN_NAME", fqdn)
-        fqdnDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
-        fqdnDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
-      }
-    }
-
-    fqdnDoc
-  }
-
-  private def getVertexIpRow(joinRow: (String, (Option[BaseDocument], Option[Row]))): BaseDocument = {
+  private def getVertexIpRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
     val ipDocOpt = joinRow._2._1
     var ipDoc = ipDocOpt match {
       case Some(doc) => doc
       case None => null
     }
-
-    val ipRowOpt = joinRow._2._2
-
-    val ipRow = ipRowOpt match {
-      case Some(r) => r
-      case None => null
-    }
-
-    if (ipRow != null){
+    val ipRow = joinRow._2._2
+    if (ipRow != null) {
       val ip = ipRow.getAs[String]("IP")
       val firstFoundTime = ipRow.getAs[Long]("FIRST_FOUND_TIME")
       val lastFoundTime = ipRow.getAs[Long]("LAST_FOUND_TIME")
@@ -228,6 +208,7 @@ object UpdateDocument {
       val ipTypeList = ipRow.getAs[ofRef[String]]("ip_type_list")
       val linkInfo = ipRow.getAs[String]("common_link_info")
       val sepAttributeTuple = separateAttributeByIpType(ipTypeList, sessionCountList, bytesSumList)
+      val vsysId = ipRow.getAs[Integer]("VSYS_ID").toLong
 
       if (ipDoc != null) {
         updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME")
@@ -235,10 +216,11 @@ object UpdateDocument {
         updateSumAttribute(ipDoc, sepAttributeTuple._2, "SERVER_BYTES_SUM")
         updateSumAttribute(ipDoc, sepAttributeTuple._3, "CLIENT_SESSION_COUNT")
         updateSumAttribute(ipDoc, sepAttributeTuple._4, "CLIENT_BYTES_SUM")
-        replaceAttribute(ipDoc,linkInfo,"COMMON_LINK_INFO")
+        replaceAttribute(ipDoc, linkInfo, "COMMON_LINK_INFO")
+        ipDoc.addAttribute("VSYS_ID", vsysId)
       } else {
         ipDoc = new BaseDocument
-        ipDoc.setKey(ip)
+        ipDoc.setKey(ip + "-" + vsysId)
         ipDoc.addAttribute("IP", ip)
         ipDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
         ipDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
@@ -246,62 +228,59 @@ object UpdateDocument {
         ipDoc.addAttribute("SERVER_BYTES_SUM", sepAttributeTuple._2)
         ipDoc.addAttribute("CLIENT_SESSION_COUNT", sepAttributeTuple._3)
         ipDoc.addAttribute("CLIENT_BYTES_SUM", sepAttributeTuple._4)
-        ipDoc.addAttribute("COMMON_LINK_INFO", "")
+        ipDoc.addAttribute("COMMON_LINK_INFO", linkInfo)
+        ipDoc.addAttribute("VSYS_ID", vsysId)
       }
     }
-
     ipDoc
   }
 
-  private def getRelationFqdnLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Option[Row]))): BaseEdgeDocument = {
+  private def getRelationFqdnLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Row))): BaseEdgeDocument = {
 
     val fqdnLocIpDocOpt = joinRow._2._1
     var fqdnLocIpDoc = fqdnLocIpDocOpt match {
       case Some(doc) => doc
       case None => null
     }
-
-    val fqdnLocIpRowOpt = joinRow._2._2
-
-    val fqdnLocIpRow = fqdnLocIpRowOpt match {
-      case Some(r) => r
-      case None => null
-    }
-
-    if (fqdnLocIpDoc != null){
+    val fqdnLocIpRow = joinRow._2._2
+    if (fqdnLocIpDoc != null) {
       updateProtocolDocument(fqdnLocIpDoc)
     }
-
-    if (fqdnLocIpRow != null){
+    if (fqdnLocIpRow != null) {
       val fqdn = fqdnLocIpRow.getAs[String]("FQDN")
-      val serverIp = fqdnLocIpRow.getAs[String]("common_server_ip")
+      val serverIp = fqdnLocIpRow.getAs[String]("server_ip")
       val firstFoundTime = fqdnLocIpRow.getAs[Long]("FIRST_FOUND_TIME")
       val lastFoundTime = fqdnLocIpRow.getAs[Long]("LAST_FOUND_TIME")
       val countTotalList = fqdnLocIpRow.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST")
-      val schemaTypeList = fqdnLocIpRow.getAs[ofRef[AnyRef]]("schema_type_list")
+      val schemaTypeList = fqdnLocIpRow.getAs[ofRef[AnyRef]]("decoded_as_list")
       val distCipRecent = fqdnLocIpRow.getAs[ofRef[String]]("DIST_CIP_RECENT")
+      val vsysId = fqdnLocIpRow.getAs[Integer]("VSYS_ID").toLong
 
       val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
       val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
-
-      val key = fqdn.concat("-" + serverIp)
-
+      val key = fqdn.concat("-" + serverIp + "-" + vsysId)
       if (fqdnLocIpDoc != null) {
+        fqdnLocIpDoc.addAttribute("FQDN", fqdn)
+        fqdnLocIpDoc.addAttribute("IP", serverIp)
+        fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
         updateMaxAttribute(fqdnLocIpDoc, lastFoundTime, "LAST_FOUND_TIME")
         updateProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
         updateDistinctIp(fqdnLocIpDoc, distinctIp)
       } else {
         fqdnLocIpDoc = new BaseEdgeDocument()
         fqdnLocIpDoc.setKey(key)
-        fqdnLocIpDoc.setFrom("FQDN/" + fqdn)
-        fqdnLocIpDoc.setTo("IP/" + serverIp)
+        fqdnLocIpDoc.setFrom("FQDN/" + fqdn+ "-" + vsysId)
+        fqdnLocIpDoc.setTo("IP/" + serverIp+ "-" + vsysId)
         fqdnLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
         fqdnLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
+        fqdnLocIpDoc.addAttribute("FQDN", fqdn)
+        fqdnLocIpDoc.addAttribute("IP", serverIp)
+        fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
+
         putProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
         putDistinctIp(fqdnLocIpDoc, distinctIp)
       }
     }
-
     fqdnLocIpDoc
   }
 

ip-learning-spark/src/main/scala/cn/ac/iie/spark/rdd/ArangoRdd.scala

@@ -1,11 +1,15 @@
 package cn.ac.iie.spark.rdd
 
+import java.util
+
 import scala.collection.JavaConverters.asScalaIteratorConverter
 import cn.ac.iie.config.ApplicationConfig
 import cn.ac.iie.service.update.UpdateDocument
 import cn.ac.iie.spark
 import cn.ac.iie.spark.partition.QueryArangoPartition
 import com.arangodb.ArangoCursor
+import com.arangodb.model.AqlQueryOptions
+import com.arangodb.util.MapBuilder
 import org.apache.spark.{Partition, SparkContext, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.slf4j.LoggerFactory
@@ -38,13 +42,15 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
 
     var arangoCursor:ArangoCursor[T] = null
     val arangoDB = spark.createArangoBuilder(options).build()
+    val bindVars: util.Map[String, AnyRef] = new MapBuilder().get
+    val queryOptions: AqlQueryOptions = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL)
     try {
       val offset = split.offset
       val separate = split.separate
       val collection = options.collection
       val sql = s"FOR doc IN $collection limit $offset,$separate RETURN doc"
       LOG.info(sql)
-      arangoCursor = arangoDB.db(options.database).query(sql,clazz.runtimeClass.asInstanceOf[Class[T]])
+      arangoCursor = arangoDB.db(options.database).query(sql,bindVars,queryOptions,clazz.runtimeClass.asInstanceOf[Class[T]])
     }catch {
       case e: Exception => LOG.error(s"创建Cursor异常:${e.getMessage}")
     }finally {
@@ -71,6 +77,9 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
     try {
       val longs = arangoDB.db(options.database).query(sql, classOf[Long])
       while (longs.hasNext) cnt = longs.next
+      if (cnt > ApplicationConfig.ARANGODB_TOTAL_NUM){
+        cnt = ApplicationConfig.ARANGODB_TOTAL_NUM
+      }
     } catch {
       case e: Exception => LOG.error(sql + s"执行异常:${e.getMessage}")
     }finally {

ip-learning-spark/src/main/scala/cn/ac/iie/utils/SparkSessionUtil.scala

@@ -21,16 +21,19 @@ object SparkSessionUtil {
       .config("spark.network.timeout", ApplicationConfig.SPARK_NETWORK_TIMEOUT)
       .config("spark.sql.shuffle.partitions", ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
       .config("spark.executor.memory", ApplicationConfig.SPARK_EXECUTOR_MEMORY)
+      .config("spark.executor.cores",ApplicationConfig.SPARK_EXECUTOR_CORES)
+      .config("spark.cores.max",ApplicationConfig.SPARK_CORES_MAX)
+      .config("spark.local.dir",ApplicationConfig.SPARK_LOCAL_DIR)
       .config("arangodb.hosts", s"${ApplicationConfig.ARANGODB_HOST}:${ApplicationConfig.ARANGODB_PORT}")
       .config("arangodb.user", ApplicationConfig.ARANGODB_USER)
       .config("arangodb.password", ApplicationConfig.ARANGODB_PASSWORD)
       .master(ApplicationConfig.MASTER)
       .getOrCreate()
-    LOG.warn("sparkession获取成功！！！")
+    LOG.warn("spark session start success！！！")
     spark
   }
 
-  def getContext: SparkContext = {
+  private def getContext: SparkContext = {
     @transient var sc: SparkContext = null
       if (sparkContext == null) sc = spark.sparkContext
     sc

ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala

@@ -15,19 +15,19 @@ object BaseClickhouseDataTest {
         |FROM
         | (
         |  (SELECT
-        |    ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
+        |    ssl_sni AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME
         |   FROM
         |    global_temp.dbtable
         |   WHERE
-        |    common_schema_type = 'SSL' GROUP BY ssl_sni
+        |    decoded_as = 'SSL' GROUP BY ssl_sni
         |  )
         |  UNION ALL
         |  (SELECT
-        |    http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
+        |    http_host AS FQDN,MAX( recv_time ) AS LAST_FOUND_TIME,MIN( recv_time ) AS FIRST_FOUND_TIME
         |   FROM
         |    global_temp.dbtable
         |   WHERE
-        |    common_schema_type = 'HTTP' GROUP BY http_host
+        |    decoded_as = 'HTTP' GROUP BY http_host
         |  )
         | )
         |GROUP BY