[修复] IP Learning 适配 Client/Server字段重命名的变化

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,30 +256,30 @@ 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.session_record 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.session_record 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_c2s) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP";
-        String serverIpSql = "SELECT common_server_ip AS IP, MIN(common_recv_time) AS FIRST_FOUND_TIME,MAX(common_recv_time) AS LAST_FOUND_TIME,count(*) as SESSION_COUNT,sum(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,groupUniqArray(2)(common_link_info_s2c) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.session_record 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.session_record 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.session_record 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.session_record 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.session_record 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 != ''";
     }
 

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

@@ -20,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-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java

@@ -83,8 +83,7 @@ public class ArangoDBConnect {
                 collection.replaceDocuments(docUpdate);
             }
         } catch (Exception e) {
-            System.out.println("更新失败");
-            e.printStackTrace();
+            LOG.error("update failure: " + e.toString());
         } finally {
             docInsert.clear();
             docInsert.clear();
@@ -102,11 +101,11 @@ public class ArangoDBConnect {
                 MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
                 Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
                 for (ErrorEntity errorEntity : errors) {
-                    LOG.warn("写入arangoDB异常：" + errorEntity.getErrorMessage());
+                    LOG.warn("write arangoDB error: " + errorEntity.getErrorMessage());
                 }
             }
         } catch (Exception e) {
-            LOG.error("更新失败：" + e.toString());
+            LOG.error("update failure: " + e.toString());
         } finally {
             docOverwrite.clear();
         }

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

@@ -3,6 +3,7 @@ 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
@@ -19,7 +20,7 @@ spark.read.clickhouse.session.table=session_record
 spark.read.clickhouse.radius.table=radius_record
 clickhouse.socket.timeout=300000
 #arangoDB配置
-arangoDB.host=192.168.44.83
+arangoDB.host=192.168.44.12
 arangoDB.port=8529
 arangoDB.user=root
 arangoDB.password=galaxy_2019
@@ -29,7 +30,7 @@ arangoDB.ttl=3600
 thread.pool.number=10
 
 #读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
-clickhouse.time.limit.type=1
+clickhouse.time.limit.type=0
 read.clickhouse.max.time=1634902508
 read.clickhouse.min.time=1631759985
 
@@ -42,4 +43,3 @@ update.interval=3600
 arangodb.total.num=20000000
 #读取radius时间范围,与radius任务执行周期一致,单位:分钟
 read.radius.granularity=-30
-vsys.id=1

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

@@ -14,6 +14,7 @@ object ApplicationConfig {
 //  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")

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

@@ -15,7 +15,7 @@ object BaseArangoData {
   def loadArangoRdd[T: ClassTag](name:String): ArangoRdd[T] ={
     val value = ArangoSpark.load[T](sparkContext, name, options)
 
-    LOG.warn(s"读取$name arangoDb:${value.count()}")
+    LOG.warn(s"read $name arangoDb: ${value.count()}")
     value
   }
 

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

@@ -1,13 +1,13 @@
 package cn.ac.iie.dao
 
-import java.util.Date
-
 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)
 
@@ -36,21 +36,21 @@ object BaseClickhouseData {
   }
 
   def getVertexFqdnDf: DataFrame = {
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+    val where = "recv_time >= " + timeLimit._2 + " AND recv_time < " + timeLimit._1
     val sql =
       s"""
          |(SELECT
-         | FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
+         | 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( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
+         |    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 common_schema_type = 'SSL' GROUP BY ssl_sni,common_vsys_id
+         |   WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,vsys_id
          |  )UNION ALL
          |  (SELECT
-         |    http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
+         |    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 common_schema_type = 'HTTP' GROUP BY http_host,common_vsys_id))
+         |   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)
@@ -60,32 +60,32 @@ object BaseClickhouseData {
   }
 
   def getVertexIpDf: DataFrame = {
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+    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
-         |,common_vsys_id AS VSYS_ID
+         |,vsys_id AS VSYS_ID
          |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
          |where $where
-         |group by common_client_ip,common_vsys_id)
+         |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
-         |,common_vsys_id AS VSYS_ID
+         |,vsys_id AS VSYS_ID
          |FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
          |where $where
-         |group by common_server_ip,common_vsys_id))) as dbtable
+         |group by server_ip,vsys_id))) as dbtable
       """.stripMargin
     LOG.warn(sql)
     val frame = initClickhouseData(sql)
@@ -95,20 +95,20 @@ object BaseClickhouseData {
 
 
   def getRelationFqdnLocateIpDf: DataFrame = {
-    val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+    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,common_vsys_id AS VSYS_ID
+         |(SELECT ssl_sni AS FQDN,destination_ip AS 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})(source_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 common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip,common_vsys_id)
+         |WHERE $where and decoded_as = 'SSL' and notEmpty(ssl_sni) and notEmpty(destination_ip) and vsys_id IS NOT NULL GROUP BY ssl_sni,destination_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,common_vsys_id AS VSYS_ID
+         |(SELECT http_host AS FQDN,destination_ip AS 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})(source_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 common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip,common_vsys_id))
-         |WHERE FQDN != '') as dbtable
+         |WHERE $where and decoded_as = 'HTTP' and notEmpty(http_host) and notEmpty(destination_ip) and vsys_id IS NOT NULL GROUP BY http_host,destination_ip,vsys_id)
+         |) as dbtable
       """.stripMargin
     LOG.warn(sql)
     val frame = initClickhouseData(sql)

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

@@ -2,6 +2,7 @@ package cn.ac.iie.main
 
 import cn.ac.iie.service.update.UpdateDocument
 
+@Deprecated
 object IpRecommendApplication {
 
   def main(args: Array[String]): Unit = {

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

@@ -2,6 +2,7 @@ package cn.ac.iie.main
 
 import cn.ac.iie.service.update.UpdateDocument
 
+@Deprecated
 object SubscriberRecommendApplication {
 
   def main(args: Array[String]): Unit = {

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

@@ -39,7 +39,7 @@ object MergeDataFrame {
     )
 
     val ipRddRow = frame.rdd.map(row => {
-      val vsysId = row.getAs[Long]("VSYS_ID")
+      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))*/
@@ -50,22 +50,26 @@ object MergeDataFrame {
 
   def mergeRelationFqdnLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
     val frame = BaseClickhouseData.getRelationFqdnLocateIpDf
-      .repartition().filter(row => isDomain(row.getAs[String]("FQDN")))
-      .groupBy("FQDN", "common_server_ip", "VSYS_ID")
+      .repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
+      .filter(row => isDomain(row.getAs[String]("FQDN")) &&
+        row.getAs[String]("server_ip") != null &&
+        row.getAs[Integer]("VSYS_ID") != null)
+      .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 vsysId = row.getAs[Long]("VSYS_ID")
-      val key = fqdn.concat("-" + serverIp + "-" + vsysId)
-      (key, row)
-    }) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
+        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)
 
@@ -133,7 +137,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/UpdateDocument.scala

@@ -72,26 +72,25 @@ object UpdateDocument {
           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"更新$collName 条数:${fqdnAccmu.value}")
+      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}ms")
     } catch {
       case e: Exception => e.printStackTrace()
     }
@@ -107,13 +106,13 @@ 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)
       }
     })
   }
@@ -192,34 +191,6 @@ object UpdateDocument {
     subidDoc
   }
 
-  private def getVertexFqdnRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
-    val fqdnDocOpt = joinRow._2._1
-    var fqdnDoc = fqdnDocOpt match {
-      case Some(doc) => doc
-      case None => null
-    }
-    val fqdnRow: Row = joinRow._2._2
-    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")
-      val vsysId = fqdnRow.getAs[Long]("VSYS_ID")
-
-      if (fqdnDoc != null) {
-        updateMaxAttribute(fqdnDoc, lastFoundTime, "LAST_FOUND_TIME")
-        fqdnDoc.addAttribute("VSYS_ID", vsysId)
-      } else {
-        fqdnDoc = new BaseDocument
-        fqdnDoc.setKey(fqdn + "-" + vsysId)
-        fqdnDoc.addAttribute("FQDN_NAME", fqdn)
-        fqdnDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
-        fqdnDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
-        fqdnDoc.addAttribute("VSYS_ID", vsysId)
-      }
-    }
-    fqdnDoc
-  }
-
   private def getVertexIpRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
     val ipDocOpt = joinRow._2._1
     var ipDoc = ipDocOpt match {
@@ -236,7 +207,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[Long]("VSYS_ID")
+      val vsysId = ipRow.getAs[Integer]("VSYS_ID").toLong
 
       if (ipDoc != null) {
         updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME")
@@ -276,13 +247,13 @@ object UpdateDocument {
     }
     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[Long]("VSYS_ID")
+      val vsysId = fqdnLocIpRow.getAs[Integer]("VSYS_ID").toLong
 
       val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
       val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)

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

@@ -49,10 +49,10 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
       val separate = split.separate
       val collection = options.collection
       val sql = s"FOR doc IN $collection limit $offset,$separate RETURN doc"
-      LOG.info(sql)
+      LOG.info(s"Executing query: $sql")
       arangoCursor = arangoDB.db(options.database).query(sql,bindVars,queryOptions,clazz.runtimeClass.asInstanceOf[Class[T]])
     }catch {
-      case e: Exception => LOG.error(s"创建Cursor异常:${e.getMessage}")
+      case e: Exception => LOG.error(s"create Cursor error: ${e.getMessage}")
     }finally {
       arangoDB.shutdown()
     }
@@ -81,7 +81,7 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
         cnt = ApplicationConfig.ARANGODB_TOTAL_NUM
       }
     } catch {
-      case e: Exception => LOG.error(sql + s"执行异常:${e.getMessage}")
+      case e: Exception => LOG.error(sql + s"execute error: ${e.getMessage}")
     }finally {
       arangoDB.shutdown()
     }

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

@@ -23,12 +23,14 @@ object SparkSessionUtil {
       .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)
+      .config("spark.sql.objectHashAggregate.sortBased.fallbackThreshold", Integer.MAX_VALUE)
       .master(ApplicationConfig.MASTER)
       .getOrCreate()
-    LOG.warn("sparkession获取成功！！！")
+    LOG.warn("spark session start success！！！")
     spark
   }
 

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