[修复] 修复iplearning查询clickhouse的SQL别名与字段相同引起的查询SSL数据为空问题 (TSG-22783)

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/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/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  decoded_as,vsys_id AS VSYS_ID
+         |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,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,vsys_id AS VSYS_ID
+         |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

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

@@ -56,7 +56,7 @@ object MergeDataFrame {
         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("decoded_as").alias("decoded_as_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 => {

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

@@ -23,6 +23,7 @@ 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)