Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
055d3bfec2 | ||
|
|
bf707f8b2e |
@@ -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>
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
#spark任务配置
|
||||
spark.sql.shuffle.partitions=10
|
||||
spark.executor.memory=4g
|
||||
spark.executor.cores=1
|
||||
spark.cores.max=10
|
||||
spark.app.name=test
|
||||
spark.network.timeout=300s
|
||||
spark.serializer=org.apache.spark.serializer.KryoSerializer
|
||||
@@ -17,22 +19,19 @@ spark.read.clickhouse.session.table=session_record
|
||||
spark.read.clickhouse.radius.table=radius_record
|
||||
clickhouse.socket.timeout=300000
|
||||
#arangoDB配置
|
||||
#arangoDB.host=192.168.40.223
|
||||
arangoDB.host=192.168.44.12
|
||||
arangoDB.port=8529
|
||||
arangoDB.user=root
|
||||
#arangoDB.password=galaxy_2019
|
||||
arangoDB.password=ceiec2019
|
||||
arangoDB.DB.name=tsg_galaxy_v3_test
|
||||
#arangoDB.DB.name=iplearn_media_domain
|
||||
arangoDB.ttl=3600
|
||||
|
||||
thread.pool.number=10
|
||||
|
||||
#读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
|
||||
clickhouse.time.limit.type=0
|
||||
read.clickhouse.max.time=1608518990
|
||||
read.clickhouse.min.time=1604851201
|
||||
clickhouse.time.limit.type=1
|
||||
read.clickhouse.max.time=1634902508
|
||||
read.clickhouse.min.time=1631759985
|
||||
|
||||
update.arango.batch=10000
|
||||
|
||||
@@ -40,5 +39,6 @@ distinct.client.ip.num=10000
|
||||
recent.count.hour=24
|
||||
|
||||
update.interval=3600
|
||||
|
||||
arangodb.total.num=20000000
|
||||
#读取radius时间范围,与radius任务执行周期一致,单位:分钟
|
||||
read.radius.granularity=-30
|
||||
|
||||
@@ -7,6 +7,8 @@ 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")
|
||||
@@ -46,4 +48,6 @@ object ApplicationConfig {
|
||||
|
||||
val ARANGODB_TOTAL_NUM: Long = config.getLong("arangodb.total.num")
|
||||
|
||||
val READ_RADIUS_GRANULARITY: Int = config.getInt("read.radius.granularity")
|
||||
|
||||
}
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
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
|
||||
|
||||
@@ -36,19 +39,19 @@ object BaseClickhouseData {
|
||||
val where = "common_recv_time >= " + timeLimit._2 + " AND common_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
|
||||
|FROM
|
||||
| ((SELECT
|
||||
| ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME
|
||||
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
| WHERE $where and 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 ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
| WHERE $where and 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,common_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
|
||||
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
| WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_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
|
||||
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
| WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_vsys_id))
|
||||
|GROUP BY FQDN,VSYS_ID HAVING FQDN != '') as dbtable
|
||||
""".stripMargin
|
||||
LOG.warn(sql)
|
||||
val frame = initClickhouseData(sql)
|
||||
@@ -67,9 +70,10 @@ object BaseClickhouseData {
|
||||
|SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
|
||||
|groupUniqArray(2)(common_link_info_c2s)[2] as common_link_info,
|
||||
|'client' as ip_type
|
||||
|,common_vsys_id AS VSYS_ID
|
||||
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
|where $where
|
||||
|group by common_client_ip)
|
||||
|group by common_client_ip,common_vsys_id)
|
||||
|UNION ALL
|
||||
|(SELECT common_server_ip AS IP,
|
||||
|MIN(common_recv_time) AS FIRST_FOUND_TIME,
|
||||
@@ -78,9 +82,10 @@ object BaseClickhouseData {
|
||||
|SUM(common_c2s_byte_num+common_s2c_byte_num) as BYTES_SUM,
|
||||
|groupUniqArray(2)(common_link_info_s2c)[2] as common_link_info,
|
||||
|'server' as ip_type
|
||||
|,common_vsys_id AS VSYS_ID
|
||||
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|
||||
|where $where
|
||||
|group by common_server_ip))) as dbtable
|
||||
|group by common_server_ip,common_vsys_id))) as dbtable
|
||||
""".stripMargin
|
||||
LOG.warn(sql)
|
||||
val frame = initClickhouseData(sql)
|
||||
@@ -95,14 +100,14 @@ object BaseClickhouseData {
|
||||
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
|
||||
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'TLS' AS schema_type,common_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)
|
||||
|WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip,common_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
|
||||
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'HTTP' AS schema_type,common_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))
|
||||
|WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip,common_vsys_id))
|
||||
|WHERE FQDN != '') as dbtable
|
||||
""".stripMargin
|
||||
LOG.warn(sql)
|
||||
@@ -114,17 +119,19 @@ object BaseClickhouseData {
|
||||
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
|
||||
|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
|
||||
|WHERE $where GROUP BY common_subscriber_id,radius_framed_ip,common_vsys_id
|
||||
|) as dbtable
|
||||
""".stripMargin
|
||||
LOG.warn(sql)
|
||||
@@ -136,8 +143,8 @@ object BaseClickhouseData {
|
||||
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
|
||||
@@ -146,8 +153,8 @@ object BaseClickhouseData {
|
||||
val sql =
|
||||
s"""
|
||||
|(
|
||||
|SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_RADIUS_TABLE}
|
||||
|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)
|
||||
@@ -159,19 +166,18 @@ object BaseClickhouseData {
|
||||
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 radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_RADIUS_TABLE} WHERE $where
|
||||
|GROUP BY radius_framed_ip
|
||||
|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)
|
||||
@@ -180,6 +186,17 @@ 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)
|
||||
}
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
println(getRadiusTimeRange)
|
||||
println(getRadiusTimeRange._2 - getRadiusTimeRange._1)
|
||||
}
|
||||
|
||||
private def getTimeLimit: (Long, Long) = {
|
||||
var maxTime = 0L
|
||||
|
||||
@@ -5,6 +5,6 @@ import cn.ac.iie.service.update.UpdateDocument
|
||||
object IpLearningApplication {
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
UpdateDocument.update()
|
||||
UpdateDocument.ipLearning()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
package cn.ac.iie.main
|
||||
|
||||
import cn.ac.iie.service.update.UpdateDocument
|
||||
|
||||
object IpRecommendApplication {
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
UpdateDocument.ipRecommend()
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package cn.ac.iie.main
|
||||
|
||||
import cn.ac.iie.service.update.UpdateDocument
|
||||
|
||||
object SubscriberRecommendApplication {
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
UpdateDocument.subscriberRecommend()
|
||||
}
|
||||
|
||||
}
|
||||
@@ -29,13 +29,14 @@ object MergeDataFrame {
|
||||
|
||||
def mergeVertexIp(): RDD[(String, (Option[BaseDocument], Row))] = {
|
||||
val vertexIpDf = BaseClickhouseData.getVertexIpDf
|
||||
val frame = vertexIpDf.repartition().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"),
|
||||
collect_list("BYTES_SUM").alias("BYTES_SUM_LIST"),
|
||||
collect_list("ip_type").alias("ip_type_list"),
|
||||
last("common_link_info").alias("common_link_info")
|
||||
last("common_link_info").alias("common_link_info"),
|
||||
last("VSYS_ID").alias("VSYS_ID")
|
||||
)
|
||||
val ipRddRow = frame.rdd.map(row => {
|
||||
(row.getAs[String]("IP"), row)
|
||||
@@ -48,13 +49,14 @@ 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")
|
||||
.groupBy("FQDN", "common_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_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT")
|
||||
collect_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT"),
|
||||
last("VSYS_ID").alias("VSYS_ID")
|
||||
)
|
||||
val fqdnLocIpRddRow = frame.rdd.map(row => {
|
||||
val fqdn = row.getAs[String]("FQDN")
|
||||
|
||||
@@ -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 {
|
||||
@@ -57,7 +72,7 @@ object UpdateDocument {
|
||||
val document: T = getDocumentRow(row)
|
||||
if (document != null) {
|
||||
fqdnAccmu.add(1)
|
||||
|
||||
// println(document)
|
||||
resultDocumentList.add(document)
|
||||
}
|
||||
i += 1
|
||||
@@ -118,23 +133,18 @@ object UpdateDocument {
|
||||
case Some(doc) => doc
|
||||
case None => null
|
||||
}
|
||||
|
||||
val subidLocIpRow = joinRow._2._2
|
||||
|
||||
// val subidLocIpRow = subidLocIpRowOpt match {
|
||||
// case Some(r) => r
|
||||
// case None => null
|
||||
// }
|
||||
|
||||
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")
|
||||
subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
|
||||
} else {
|
||||
subidLocIpDoc = new BaseEdgeDocument()
|
||||
subidLocIpDoc.setKey(key)
|
||||
@@ -144,6 +154,7 @@ object UpdateDocument {
|
||||
subidLocIpDoc.addAttribute("IP", ip)
|
||||
subidLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
|
||||
subidLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
|
||||
subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
|
||||
}
|
||||
}
|
||||
subidLocIpDoc
|
||||
@@ -155,29 +166,25 @@ object UpdateDocument {
|
||||
case Some(doc) => doc
|
||||
case None => null
|
||||
}
|
||||
|
||||
val subidRow = joinRow._2._2
|
||||
|
||||
// val subidRow = subidRowOpt match {
|
||||
// case Some(r) => r
|
||||
// case None => null
|
||||
// }
|
||||
|
||||
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")
|
||||
val vsysId = subidRow.getAs[Long]("VSYS_ID")
|
||||
|
||||
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.addAttribute("VSYS_ID", vsysId)
|
||||
}
|
||||
}
|
||||
|
||||
subidDoc
|
||||
}
|
||||
|
||||
@@ -187,29 +194,25 @@ object UpdateDocument {
|
||||
case Some(doc) => doc
|
||||
case None => null
|
||||
}
|
||||
|
||||
val fqdnRow: Row = 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")
|
||||
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)
|
||||
fqdnDoc.addAttribute("FQDN_NAME", fqdn)
|
||||
fqdnDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
|
||||
fqdnDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
|
||||
fqdnDoc.addAttribute("VSYS_ID", vsysId)
|
||||
}
|
||||
}
|
||||
|
||||
fqdnDoc
|
||||
}
|
||||
|
||||
@@ -219,14 +222,7 @@ object UpdateDocument {
|
||||
case Some(doc) => doc
|
||||
case None => null
|
||||
}
|
||||
|
||||
val ipRow = joinRow._2._2
|
||||
|
||||
// val ipRow = ipRowOpt match {
|
||||
// case Some(r) => r
|
||||
// case None => null
|
||||
// }
|
||||
|
||||
if (ipRow != null) {
|
||||
val ip = ipRow.getAs[String]("IP")
|
||||
val firstFoundTime = ipRow.getAs[Long]("FIRST_FOUND_TIME")
|
||||
@@ -236,6 +232,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")
|
||||
|
||||
if (ipDoc != null) {
|
||||
updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME")
|
||||
@@ -244,6 +241,7 @@ object UpdateDocument {
|
||||
updateSumAttribute(ipDoc, sepAttributeTuple._3, "CLIENT_SESSION_COUNT")
|
||||
updateSumAttribute(ipDoc, sepAttributeTuple._4, "CLIENT_BYTES_SUM")
|
||||
replaceAttribute(ipDoc, linkInfo, "COMMON_LINK_INFO")
|
||||
ipDoc.addAttribute("VSYS_ID", vsysId)
|
||||
} else {
|
||||
ipDoc = new BaseDocument
|
||||
ipDoc.setKey(ip)
|
||||
@@ -255,9 +253,9 @@ object UpdateDocument {
|
||||
ipDoc.addAttribute("CLIENT_SESSION_COUNT", sepAttributeTuple._3)
|
||||
ipDoc.addAttribute("CLIENT_BYTES_SUM", sepAttributeTuple._4)
|
||||
ipDoc.addAttribute("COMMON_LINK_INFO", "")
|
||||
ipDoc.addAttribute("VSYS_ID", vsysId)
|
||||
}
|
||||
}
|
||||
|
||||
ipDoc
|
||||
}
|
||||
|
||||
@@ -268,18 +266,10 @@ object UpdateDocument {
|
||||
case Some(doc) => doc
|
||||
case None => null
|
||||
}
|
||||
|
||||
val fqdnLocIpRow = joinRow._2._2
|
||||
|
||||
// val fqdnLocIpRow = fqdnLocIpRowOpt match {
|
||||
// case Some(r) => r
|
||||
// case None => null
|
||||
// }
|
||||
|
||||
if (fqdnLocIpDoc != null) {
|
||||
updateProtocolDocument(fqdnLocIpDoc)
|
||||
}
|
||||
|
||||
if (fqdnLocIpRow != null) {
|
||||
val fqdn = fqdnLocIpRow.getAs[String]("FQDN")
|
||||
val serverIp = fqdnLocIpRow.getAs[String]("common_server_ip")
|
||||
@@ -288,16 +278,16 @@ object UpdateDocument {
|
||||
val countTotalList = fqdnLocIpRow.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST")
|
||||
val schemaTypeList = fqdnLocIpRow.getAs[ofRef[AnyRef]]("schema_type_list")
|
||||
val distCipRecent = fqdnLocIpRow.getAs[ofRef[String]]("DIST_CIP_RECENT")
|
||||
val vsysId = fqdnLocIpRow.getAs[Long]("VSYS_ID")
|
||||
|
||||
val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
|
||||
val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
|
||||
|
||||
val key = fqdn.concat("-" + serverIp)
|
||||
|
||||
if (fqdnLocIpDoc != null) {
|
||||
updateMaxAttribute(fqdnLocIpDoc, lastFoundTime, "LAST_FOUND_TIME")
|
||||
updateProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
|
||||
updateDistinctIp(fqdnLocIpDoc, distinctIp)
|
||||
fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
|
||||
} else {
|
||||
fqdnLocIpDoc = new BaseEdgeDocument()
|
||||
fqdnLocIpDoc.setKey(key)
|
||||
@@ -305,11 +295,12 @@ object UpdateDocument {
|
||||
fqdnLocIpDoc.setTo("IP/" + serverIp)
|
||||
fqdnLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
|
||||
fqdnLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
|
||||
fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
|
||||
|
||||
putProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
|
||||
putDistinctIp(fqdnLocIpDoc, distinctIp)
|
||||
}
|
||||
}
|
||||
|
||||
fqdnLocIpDoc
|
||||
}
|
||||
|
||||
|
||||
@@ -21,6 +21,8 @@ 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("arangodb.hosts", s"${ApplicationConfig.ARANGODB_HOST}:${ApplicationConfig.ARANGODB_PORT}")
|
||||
.config("arangodb.user", ApplicationConfig.ARANGODB_USER)
|
||||
.config("arangodb.password", ApplicationConfig.ARANGODB_PASSWORD)
|
||||
|
||||
Reference in New Issue
Block a user