gfwleak/wanglihui-ip-learning-graph
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: gfwleak:tsg-v2
Branches Tags
gfwleak:master gfwleak:tsg-v2412 gfwleak:tsg-v2402 gfwleak:tsg-v2311 gfwleak:tsg-v2210 gfwleak:tsg-v2110 gfwleak:tsg-v2 gfwleak:ip-learning-graph-tsg gfwleak:ip-learning-graph-report gfwleak:ip-learning-graph-zx gfwleak:ip-learning-graph-YSP
...
pull from: gfwleak:tsg-v2311
Branches Tags
gfwleak:tsg-v2412 gfwleak:tsg-v2402 gfwleak:tsg-v2311 gfwleak:tsg-v2210 gfwleak:tsg-v2110 gfwleak:tsg-v2 gfwleak:master gfwleak:ip-learning-graph-tsg gfwleak:ip-learning-graph-report gfwleak:ip-learning-graph-zx gfwleak:ip-learning-graph-YSP

11 Commits
tsg-v2 ... tsg-v2311

Author SHA1 Message Date
zhanghongqing
bdf8b00285 处理VSYS_ID类型转换异常 2023-11-13 18:43:26 +08:00
zhanghongqing
219e66cea1 修改spark离线任务，IP-Learning 相关学习功能，WAN NAT Pools功能（暂时停止任务） TSG-17496 2023-11-02 18:04:32 +08:00
zhanghongqing
272df41aec 修改ip池数据linkInfo设置方式 2022-10-24 14:37:07 +08:00
zhanghongqing
b8601cc26e 去掉 System.exit(0) 退出 2022-09-23 11:29:04 +08:00
zhanghongqing
4937d33e3e 修复iplearning 补全LAST_FOUND_TIME 2022-09-22 09:46:33 +08:00
zhanghongqing
110ad435f1 R_LOCATE_FQDN2IP增加字段 FQDN，IP 2022-09-20 18:14:48 +08:00
zhanghongqing
e4260517fc SUBSCRIBER from to 增加vsys_id 2022-09-20 16:36:44 +08:00
zhanghongqing
1bf1045223 TSG-11902 2022-09-20 11:08:54 +08:00
zhanghongqing
4bc5b75994 增加vsys_id字段 TSG-11902 2022-09-19 16:21:33 +08:00
zhanghongqing
055d3bfec2 增加vsys_id字段 TSG-11902 2022-09-19 10:05:51 +08:00
wanglihui
bf707f8b2e 资源隔离三个任务，增加spark.executor.cores、spark.cores.max资源控制参数。 2021-10-29 18:54:18 +08:00
12 changed files with 447 additions and 260 deletions
Expand all files Collapse all files

32
IP-learning-graph/src/main/java/cn/ac/iie/service/ingestion/ReadClickhouseData.java
View File

@@ -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 != ''";
}

320
ip-learning-spark/pom.xml
View File

@@ -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>

27
ip-learning-spark/src/main/resources/application.properties
View File

@@ -1,38 +1,37 @@
#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
master=local[*]
#spark读取clickhouse配置
spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.67:8123/tsg_galaxy_v3
spark.read.clickhouse.url=jdbc:clickhouse://192.168.44.12:8123/tsg_galaxy_v3
spark.read.clickhouse.driver=ru.yandex.clickhouse.ClickHouseDriver
spark.read.clickhouse.user=default
spark.read.clickhouse.password=ceiec2019
spark.read.clickhouse.password=galaxy2019
spark.read.clickhouse.numPartitions=5
spark.read.clickhouse.fetchsize=10000
spark.read.clickhouse.fetchsize=10
spark.read.clickhouse.partitionColumn=LAST_FOUND_TIME
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.host=192.168.44.83
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.password=galaxy_2019
arangoDB.DB.name=tsg_galaxy_v3
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,7 @@ distinct.client.ip.num=10000
recent.count.hour=24
update.interval=3600
arangodb.total.num=20000000
#读取radius时间范围,与radius任务执行周期一致,单位:分钟
read.radius.granularity=-30
vsys.id=1

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

@@ -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")
}

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

@@ -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
@@ -33,22 +36,22 @@ 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
|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,vsys_id AS VSYS_ID
|FROM
| ((SELECT
| 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 decoded_as = 'SSL' GROUP BY ssl_sni,vsys_id
| )UNION ALL
| (SELECT
| 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 decoded_as = 'HTTP' GROUP BY http_host,vsys_id))
|GROUP BY FQDN,VSYS_ID HAVING FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
@@ -57,30 +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
|,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_client_ip)
|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
|,vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_server_ip))) as dbtable
|group by server_ip,vsys_id))) as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
@@ -90,19 +95,19 @@ 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
|((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
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip)
|WHERE $where and decoded_as = 'SSL' GROUP BY ssl_sni,server_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
|(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
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip))
|WHERE $where and decoded_as = 'HTTP' GROUP BY http_host,server_ip,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,18 +143,19 @@ 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
| AND radius_acct_status_type = 1
""".stripMargin
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 +167,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 +187,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

2
ip-learning-spark/src/main/scala/cn/ac/iie/main/IpLearningApplication.scala
View File

@@ -5,6 +5,6 @@ import cn.ac.iie.service.update.UpdateDocument
object IpLearningApplication {
def main(args: Array[String]): Unit = {
UpdateDocument.update()
UpdateDocument.ipLearning()
}
}

11
ip-learning-spark/src/main/scala/cn/ac/iie/main/IpRecommendApplication.scala Normal file
View File

@@ -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()
}
}

11
ip-learning-spark/src/main/scala/cn/ac/iie/main/SubscriberRecommendApplication.scala Normal file
View File

@@ -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()
}
}

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

@@ -1,7 +1,5 @@
package cn.ac.iie.service.transform
import java.util.regex.Pattern
import cn.ac.iie.config.ApplicationConfig
import cn.ac.iie.dao.{BaseArangoData, BaseClickhouseData}
import cn.ac.iie.spark.partition.CustomPartitioner
@@ -12,15 +10,17 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.functions._
import org.slf4j.LoggerFactory
import java.util.regex.Pattern
object MergeDataFrame {
private val LOG = LoggerFactory.getLogger(MergeDataFrame.getClass)
private val pattern = Pattern.compile("^[\\d]*$")
def mergeVertexFqdn(): RDD[(String, (Option[BaseDocument], Row))] = {
val fqdnRddRow: RDD[(String, Row)] = BaseClickhouseData.getVertexFqdnDf
.repartition().rdd.filter(row => isDomain(row.getAs[String](0))).map(row => {
.repartition().rdd.filter(row => isDomain(row.getAs[String](0))).map(row => {
(row.getAs[String]("FQDN"), row)
})/*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
}) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
val fqdnRddDoc: ArangoRdd[BaseDocument] = BaseArangoData.loadArangoRdd[BaseDocument]("FQDN")
@@ -29,7 +29,7 @@ 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"),
@@ -37,9 +37,12 @@ object MergeDataFrame {
collect_list("ip_type").alias("ip_type_list"),
last("common_link_info").alias("common_link_info")
)
val ipRddRow = frame.rdd.map(row => {
(row.getAs[String]("IP"), row)
})/*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
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))*/
val ipRddDoc = BaseArangoData.loadArangoRdd[BaseDocument]("IP")
ipRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(ipRddRow)
@@ -48,20 +51,21 @@ 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", "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 key = fqdn.concat("-" + serverIp)
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))*/
}) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
val fqdnLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_FQDN2IP")
fqdnLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnLocIpRddRow)
@@ -73,7 +77,8 @@ object MergeDataFrame {
.rdd.map(row => {
val commonSubscriberId = row.getAs[String]("common_subscriber_id")
val ip = row.getAs[String]("radius_framed_ip")
val key = commonSubscriberId.concat("-" + ip)
val vsysId = row.getAs[Long]("VSYS_ID")
val key = commonSubscriberId.concat("-" + ip + "-" + vsysId)
(key, row)
}).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
val subidLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_SUBSCRIBER2IP")

149
ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala
View File

@@ -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
@@ -105,8 +120,9 @@ object UpdateDocument {
private def getVertexFrameipRow(row: Row): BaseDocument = {
val ip = row.getAs[String]("radius_framed_ip")
val vsysId = row.getAs[Long]("VSYS_ID")
val document = new BaseDocument()
document.setKey(ip)
document.setKey(ip + "-" + vsysId)
document.addAttribute("IP", ip)
document
}
@@ -118,32 +134,30 @@ 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)
val key = subId.concat("-" + ip + "-" + vsysId)
if (subidLocIpDoc != null) {
subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
subidLocIpDoc.addAttribute("SUBSCRIBER", subId)
subidLocIpDoc.addAttribute("IP", ip)
updateMaxAttribute(subidLocIpDoc, lastFoundTime, "LAST_FOUND_TIME")
} else {
subidLocIpDoc = new BaseEdgeDocument()
subidLocIpDoc.setKey(key)
subidLocIpDoc.setFrom("SUBSCRIBER/" + subId)
subidLocIpDoc.setTo("IP/" + ip)
subidLocIpDoc.setFrom("SUBSCRIBER/" + subId + "-" + vsysId)
subidLocIpDoc.setTo("IP/" + ip+ "-" + vsysId)
subidLocIpDoc.addAttribute("SUBSCRIBER", subId)
subidLocIpDoc.addAttribute("IP", ip)
subidLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
subidLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
subidLocIpDoc.addAttribute("VSYS_ID", vsysId)
}
}
subidLocIpDoc
@@ -155,78 +169,36 @@ 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")
val key = subId.concat("-" + vsysId)
if (subidDoc != null) {
updateMaxAttribute(subidDoc, subLastFoundTime, "LAST_FOUND_TIME")
subidDoc.addAttribute("VSYS_ID", vsysId)
} else {
subidDoc = new BaseDocument()
subidDoc.setKey(subId)
subidDoc.setKey(key)
subidDoc.addAttribute("SUBSCRIBER", subId)
subidDoc.addAttribute("FIRST_FOUND_TIME", subFirstFoundTime)
subidDoc.addAttribute("LAST_FOUND_TIME", subLastFoundTime)
subidDoc.addAttribute("VSYS_ID", vsysId)
}
}
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
// 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")
if (fqdnDoc != null) {
updateMaxAttribute(fqdnDoc, lastFoundTime, "LAST_FOUND_TIME")
} 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
}
private def getVertexIpRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
val ipDocOpt = joinRow._2._1
var ipDoc = ipDocOpt match {
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 +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[Integer]("VSYS_ID").toLong
if (ipDoc != null) {
updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME")
@@ -244,9 +217,10 @@ 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)
ipDoc.setKey(ip + "-" + vsysId)
ipDoc.addAttribute("IP", ip)
ipDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
ipDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
@@ -254,10 +228,10 @@ object UpdateDocument {
ipDoc.addAttribute("SERVER_BYTES_SUM", sepAttributeTuple._2)
ipDoc.addAttribute("CLIENT_SESSION_COUNT", sepAttributeTuple._3)
ipDoc.addAttribute("CLIENT_BYTES_SUM", sepAttributeTuple._4)
ipDoc.addAttribute("COMMON_LINK_INFO", "")
ipDoc.addAttribute("COMMON_LINK_INFO", linkInfo)
ipDoc.addAttribute("VSYS_ID", vsysId)
}
}
ipDoc
}
@@ -268,48 +242,45 @@ 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")
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[Integer]("VSYS_ID").toLong
val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
val key = fqdn.concat("-" + serverIp)
val key = fqdn.concat("-" + serverIp + "-" + vsysId)
if (fqdnLocIpDoc != null) {
fqdnLocIpDoc.addAttribute("FQDN", fqdn)
fqdnLocIpDoc.addAttribute("IP", serverIp)
fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
updateMaxAttribute(fqdnLocIpDoc, lastFoundTime, "LAST_FOUND_TIME")
updateProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
updateDistinctIp(fqdnLocIpDoc, distinctIp)
} else {
fqdnLocIpDoc = new BaseEdgeDocument()
fqdnLocIpDoc.setKey(key)
fqdnLocIpDoc.setFrom("FQDN/" + fqdn)
fqdnLocIpDoc.setTo("IP/" + serverIp)
fqdnLocIpDoc.setFrom("FQDN/" + fqdn+ "-" + vsysId)
fqdnLocIpDoc.setTo("IP/" + serverIp+ "-" + vsysId)
fqdnLocIpDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
fqdnLocIpDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
fqdnLocIpDoc.addAttribute("FQDN", fqdn)
fqdnLocIpDoc.addAttribute("IP", serverIp)
fqdnLocIpDoc.addAttribute("VSYS_ID", vsysId)
putProtocolAttritube(fqdnLocIpDoc, sepAttritubeMap)
putDistinctIp(fqdnLocIpDoc, distinctIp)
}
}
fqdnLocIpDoc
}

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

@@ -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)

8
ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala
View File

@@ -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