diff options
7 files changed, 27 insertions, 55 deletions
diff --git a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java b/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java index 39b0439..16b79a6 100644 --- a/IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java +++ b/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); @@ -257,29 +257,29 @@ public class ReadClickhouseData { public static String getVertexFqdnSql() { 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 common_schema_type = '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 common_schema_type = 'HTTP' GROUP BY http_host"; + 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 = " recv_time >= " + minTime + " AND recv_time < " + maxTime; - String clientIpSql = "SELECT common_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) 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(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) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.session_record where " + where + " group by IP"; + 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 = " recv_time >= " + minTime + " AND recv_time < " + maxTime; - String sslSql = "SELECT ssl_sni AS FQDN,common_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+")(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(recv_time) AS LAST_FOUND_TIME,MIN(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 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 = " recv_time >= " + minTime + " AND recv_time < " + maxTime; - String httpSql = "SELECT http_host AS FQDN,common_client_ip,MAX(recv_time) AS LAST_FOUND_TIME,MIN(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(recv_time) AS LAST_FOUND_TIME,MIN(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 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 != ''"; } diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala index a6c691d..cfe8b5e 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala +++ b/ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala @@ -100,12 +100,12 @@ object BaseClickhouseData { s""" |(SELECT * FROM |((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 schema_type,vsys_id AS VSYS_ID + |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'TLS' AS decoded_as,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,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 schema_type,vsys_id AS VSYS_ID + |toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(client_ip)) AS DIST_CIP_RECENT,'HTTP' AS decoded_as,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 diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala index 99b188d..7fa6d62 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala +++ b/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala @@ -1,7 +1,7 @@ package cn.ac.iie.main import cn.ac.iie.service.update.UpdateDocument -@deprecated + object IpRecommendApplication { def main(args: Array[String]): Unit = { diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala index 3ee06b2..04cbec4 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala +++ b/ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala @@ -1,7 +1,7 @@ package cn.ac.iie.main import cn.ac.iie.service.update.UpdateDocument -@deprecated + object SubscriberRecommendApplication { def main(args: Array[String]): Unit = { diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala index 352a650..20ca1e8 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala +++ b/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))*/ @@ -51,18 +51,18 @@ 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") + .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").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 serverIp = row.getAs[String]("server_ip") + val vsysId = row.getAs[Integer]("VSYS_ID").toLong val key = fqdn.concat("-" + serverIp + "-" + vsysId) (key, row) }) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/ diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala index 2bd7732..96b3dcd 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala +++ b/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala @@ -192,34 +192,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 +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[Long]("VSYS_ID") + val vsysId = ipRow.getAs[Integer]("VSYS_ID").toLong if (ipDoc != null) { updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME") @@ -276,13 +248,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) diff --git a/ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala b/ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala index 01804eb..3fcf09f 100644 --- a/ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala +++ b/ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala @@ -19,7 +19,7 @@ object BaseClickhouseDataTest { | FROM | global_temp.dbtable | WHERE - | common_schema_type = 'SSL' GROUP BY ssl_sni + | decoded_as = 'SSL' GROUP BY ssl_sni | ) | UNION ALL | (SELECT @@ -27,7 +27,7 @@ object BaseClickhouseDataTest { | FROM | global_temp.dbtable | WHERE - | common_schema_type = 'HTTP' GROUP BY http_host + | decoded_as = 'HTTP' GROUP BY http_host | ) | ) |GROUP BY |