diff --git a/IP-learning-graph/src/main/resources/application.properties b/IP-learning-graph/src/main/resources/application.properties index 5e15915..eb19abb 100644 --- a/IP-learning-graph/src/main/resources/application.properties +++ b/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 diff --git a/ip-learning-spark/src/main/resources/application.properties b/ip-learning-spark/src/main/resources/application.properties index a27b28a..39e0e24 100644 --- a/ip-learning-spark/src/main/resources/application.properties +++ b/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 \ No newline at end of file diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala index 29eb786..cf17d27 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala +++ b/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") 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 cfe8b5e..b8e91f7 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 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 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 7fa6d62..7692371 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 @@ -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 = { 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 04cbec4..812e6d1 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 @@ -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 = { 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 20ca1e8..bbce2cf 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 @@ -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 => { diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/utils/SparkSessionUtil.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/utils/SparkSessionUtil.scala index f96862a..cce388e 100644 --- a/ip-learning-spark/src/main/scala/cn/ac/iie/utils/SparkSessionUtil.scala +++ b/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)