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 324634c..1167d7b 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
@@ -105,6 +105,7 @@ public class ReadClickhouseData {
newDoc.setKey(subscriberId);
newDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime);
newDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime);
+ newDoc.addAttribute("SUBSCRIBER_ID",subscriberId);
} catch (Exception e) {
e.printStackTrace();
}
diff --git a/IP-learning-graph/src/main/resources/application.properties b/IP-learning-graph/src/main/resources/application.properties
index d74c12e..d377e97 100644
--- a/IP-learning-graph/src/main/resources/application.properties
+++ b/IP-learning-graph/src/main/resources/application.properties
@@ -1,6 +1,6 @@
#arangoDB参数配置
-#arangoDB.host=192.168.40.182
-arangoDB.host=192.168.40.224
+arangoDB.host=192.168.40.182
+#arangoDB.host=192.168.40.224
arangoDB.port=8529
arangoDB.user=root
arangoDB.password=111111
@@ -17,9 +17,9 @@ thread.await.termination.time=10
#读取clickhouse时间范围方式，0：读取过去一小时，1：指定时间范围
-time.limit.type=0
-read.clickhouse.max.time=1595833062
-read.clickhouse.min.time=1595833060
+time.limit.type=1
+read.clickhouse.max.time=1596684142
+read.clickhouse.min.time=1596425769
update.interval=3600
distinct.client.ip.num=10000
diff --git a/IP-learning-graph/src/main/resources/clickhouse.properties b/IP-learning-graph/src/main/resources/clickhouse.properties
index 28e8e00..3b18aa4 100644
--- a/IP-learning-graph/src/main/resources/clickhouse.properties
+++ b/IP-learning-graph/src/main/resources/clickhouse.properties
@@ -1,9 +1,9 @@
drivers=ru.yandex.clickhouse.ClickHouseDriver
mdb.user=default
-#db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=3600000
-#mdb.password=111111
-db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=3600000
-mdb.password=ceiec2019
+db.id=192.168.40.186:8123/tsg_galaxy_v3?socket_timeout=300000
+mdb.password=111111
+#db.id=192.168.40.224:8123/tsg_galaxy_v3?socket_timeout=300000
+#mdb.password=ceiec2019
initialsize=1
minidle=1
maxactive=50
diff --git a/IP-learning-graph/src/test/java/cn/ac/iie/readHistoryDataTest.java b/IP-learning-graph/src/test/java/cn/ac/iie/readHistoryDataTest.java
new file mode 100644
index 0000000..56f9b50
--- /dev/null
+++ b/IP-learning-graph/src/test/java/cn/ac/iie/readHistoryDataTest.java
@@ -0,0 +1,10 @@
+package cn.ac.iie;
+
+import cn.ac.iie.dao.BaseArangoData;
+import com.arangodb.entity.BaseEdgeDocument;
+
+public class readHistoryDataTest {
+ public static void main(String[] args) {
+ BaseArangoData baseArangoData = new BaseArangoData();
+ }
+}
diff --git a/ip-learning-spark/.gitignore b/ip-learning-spark/.gitignore
index e13556e..5b9ff5c 100644
--- a/ip-learning-spark/.gitignore
+++ b/ip-learning-spark/.gitignore
@@ -3,8 +3,10 @@
### Example user template
# IntelliJ project files
-.idea
+.idea/
*.iml
target
logs/
spark-warehouse/
+src/main/java/cn/ac/iie/config/
+src/test/java/
diff --git a/ip-learning-spark/.idea/libraries/scala_sdk_2_11_7.xml b/ip-learning-spark/.idea/libraries/scala_sdk_2_11_7.xml
new file mode 100644
index 0000000..96b8d93
--- /dev/null
+++ b/ip-learning-spark/.idea/libraries/scala_sdk_2_11_7.xml
@@ -0,0 +1,162 @@
+
+
+
+ Scala_2_11
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/ip-learning-spark/src/main/java/cn/ac/iie/dao/BaseArangoData.java b/ip-learning-spark/src/main/java/cn/ac/iie/dao/BaseArangoData.java
new file mode 100644
index 0000000..a141219
--- /dev/null
+++ b/ip-learning-spark/src/main/java/cn/ac/iie/dao/BaseArangoData.java
@@ -0,0 +1,103 @@
+package cn.ac.iie.dao;
+
+import cn.ac.iie.config.ApplicationConfig;
+import cn.ac.iie.service.read.ReadHistoryArangoData;
+import cn.ac.iie.utils.ArangoDBConnect;
+import cn.ac.iie.utils.ExecutorThreadPool;
+import com.arangodb.ArangoCursor;
+import com.arangodb.entity.BaseDocument;
+import com.arangodb.entity.BaseEdgeDocument;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * 获取arangoDB历史数据
+ *
+ * @author wlh
+ */
+public class BaseArangoData {
+ private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
+
+ public static ConcurrentHashMap> historyVertexFqdnMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyVertexIpMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyVertexSubscriberMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyRelationFqdnAddressIpMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyRelationIpVisitFqdnMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyRelationFqdnSameFqdnMap = new ConcurrentHashMap<>();
+ public static ConcurrentHashMap> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>();
+
+ private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
+
+ private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
+
+ public void readHistoryData(String table,
+ ConcurrentHashMap> historyMap,
+ Class type) {
+ try {
+ LOG.warn("开始更新" + table);
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
+ historyMap.put(i, new ConcurrentHashMap<>());
+ }
+ CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER());
+ long[] timeRange = getTimeRange(table);
+ for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER(); i++) {
+ String sql = getQuerySql(timeRange, i, table);
+ ReadHistoryArangoData readHistoryArangoData = new ReadHistoryArangoData<>(arangoDBConnect, sql, historyMap, type, table, countDownLatch);
+ threadPool.executor(readHistoryArangoData);
+ }
+ countDownLatch.await();
+ long last = System.currentTimeMillis();
+ LOG.warn("读取" + table + " arangoDB 共耗时：" + (last - start));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ private long[] getTimeRange(String table) {
+ long minTime = 0L;
+ long maxTime = 0L;
+ long startTime = System.currentTimeMillis();
+ String sql = "LET doc = (FOR doc IN " + table + " RETURN doc) return {max_time:MAX(doc[*].FIRST_FOUND_TIME),min_time:MIN(doc[*].FIRST_FOUND_TIME)}";
+ switch (ApplicationConfig.ARANGO_TIME_LIMIT_TYPE()) {
+ case 0:
+ ArangoCursor timeDoc = arangoDBConnect.executorQuery(sql, BaseDocument.class);
+ try {
+ if (timeDoc != null) {
+ while (timeDoc.hasNext()) {
+ BaseDocument doc = timeDoc.next();
+ maxTime = Long.parseLong(doc.getAttribute("max_time").toString()) + ApplicationConfig.THREAD_POOL_NUMBER();
+ minTime = Long.parseLong(doc.getAttribute("min_time").toString());
+ }
+ } else {
+ LOG.warn("获取ArangoDb时间范围为空");
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ break;
+ case 1:
+ maxTime = ApplicationConfig.READ_ARANGO_MAX_TIME();
+ minTime = ApplicationConfig.READ_ARANGO_MIN_TIME();
+ break;
+ default:
+ }
+ long lastTime = System.currentTimeMillis();
+ LOG.warn(sql + "\n查询最大最小时间用时：" + (lastTime - startTime));
+ return new long[]{minTime, maxTime};
+
+ }
+
+ private String getQuerySql(long[] timeRange, int threadNumber, String table) {
+ long minTime = timeRange[0];
+ long maxTime = timeRange[1];
+ long diffTime = (maxTime - minTime) / ApplicationConfig.THREAD_POOL_NUMBER();
+ long maxThreadTime = minTime + (threadNumber + 1) * diffTime;
+ long minThreadTime = minTime + threadNumber * diffTime;
+ return "FOR doc IN " + table + " filter doc.FIRST_FOUND_TIME >= " + minThreadTime + " and doc.FIRST_FOUND_TIME <= " + maxThreadTime + " " + ApplicationConfig.ARANGODB_READ_LIMIT() + " RETURN doc";
+ }
+
+}
diff --git a/ip-learning-spark/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java b/ip-learning-spark/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java
new file mode 100644
index 0000000..1ca66d7
--- /dev/null
+++ b/ip-learning-spark/src/main/java/cn/ac/iie/service/read/ReadHistoryArangoData.java
@@ -0,0 +1,125 @@
+package cn.ac.iie.service.read;
+
+import cn.ac.iie.config.ApplicationConfig;
+import cn.ac.iie.utils.ArangoDBConnect;
+import com.arangodb.ArangoCursor;
+import com.arangodb.entity.BaseDocument;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+
+/**
+ * @author wlh
+ * 多线程全量读取arangoDb历史数据，封装到map
+ */
+public class ReadHistoryArangoData extends Thread {
+ public static long currentHour = System.currentTimeMillis() / (60 * 60 * 1000) * 60 * 60;
+ private static final Logger LOG = LoggerFactory.getLogger(ReadHistoryArangoData.class);
+ static final Integer RECENT_COUNT_HOUR = ApplicationConfig.RECENT_COUNT_HOUR();
+
+ public static final HashSet PROTOCOL_SET;
+
+ static {
+ PROTOCOL_SET = new HashSet<>();
+ PROTOCOL_SET.add("HTTP");
+ PROTOCOL_SET.add("TLS");
+ PROTOCOL_SET.add("DNS");
+ }
+
+ private ArangoDBConnect arangoConnect;
+ private String query;
+ private ConcurrentHashMap> map;
+ private Class type;
+ private String table;
+ private CountDownLatch countDownLatch;
+
+ public ReadHistoryArangoData(ArangoDBConnect arangoConnect,
+ String query,
+ ConcurrentHashMap> map,
+ Class type,
+ String table,
+ CountDownLatch countDownLatch) {
+ this.arangoConnect = arangoConnect;
+ this.query = query;
+ this.map = map;
+ this.type = type;
+ this.table = table;
+ this.countDownLatch = countDownLatch;
+ }
+
+ @Override
+ public void run() {
+ try {
+ long s = System.currentTimeMillis();
+ ArangoCursor docs = arangoConnect.executorQuery(query, type);
+ if (docs != null) {
+ List baseDocuments = docs.asListRemaining();
+ int i = 0;
+ for (T doc : baseDocuments) {
+ String key = doc.getKey();
+ switch (table) {
+ case "R_LOCATE_FQDN2IP":
+ updateProtocolDocument(doc);
+ deleteDistinctClientIpByTime(doc);
+ break;
+ case "R_VISIT_IP2FQDN":
+ updateProtocolDocument(doc);
+ break;
+ default:
+ }
+ int hashCode = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER();
+ ConcurrentHashMap tmpMap = map.get(hashCode);
+ tmpMap.put(key, doc);
+ i++;
+ }
+ long l = System.currentTimeMillis();
+ LOG.warn(query + "\n读取" + i + "条数据，运行时间：" + (l - s));
+ }
+ }catch (Exception e){
+ e.printStackTrace();
+ }finally {
+ countDownLatch.countDown();
+ LOG.warn("本线程读取完毕，剩余线程数量："+countDownLatch.getCount());
+ }
+ }
+
+ private void updateProtocolDocument(T doc) {
+ if (doc.getProperties().containsKey("PROTOCOL_TYPE")) {
+ for (String protocol : PROTOCOL_SET) {
+ String protocolRecent = protocol + "_CNT_RECENT";
+ ArrayList cntRecent = (ArrayList) doc.getAttribute(protocolRecent);
+ Long[] cntRecentsSrc = cntRecent.toArray(new Long[cntRecent.size()]);
+ Long[] cntRecentsDst = new Long[RECENT_COUNT_HOUR];
+ System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1);
+ cntRecentsDst[0] = 0L;
+ doc.addAttribute(protocolRecent, cntRecentsDst);
+ }
+ }
+ }
+
+ private void deleteDistinctClientIpByTime(T doc) {
+ ArrayList distCip = (ArrayList) doc.getAttribute("DIST_CIP");
+ ArrayList distCipTs = (ArrayList) doc.getAttribute("DIST_CIP_TS");
+ distCipTs.add(currentHour - RECENT_COUNT_HOUR * 3600);
+ Collections.sort(distCipTs);
+ int index = distCipTs.indexOf(currentHour - RECENT_COUNT_HOUR * 3600);
+ String[] distCipArr = new String[index];
+ long[] disCipTsArr = new long[index];
+ if (distCip.size() + 1 == distCipTs.size()){
+ for (int i = 0; i < index; i++) {
+ distCipArr[i] = distCip.get(i);
+ disCipTsArr[i] = distCipTs.get(i);
+ }
+ }
+ doc.updateAttribute("DIST_CIP", distCipArr);
+ doc.updateAttribute("DIST_CIP_TS", disCipTsArr);
+ }
+
+}
diff --git a/ip-learning-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java b/ip-learning-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java
new file mode 100644
index 0000000..d5fb1b8
--- /dev/null
+++ b/ip-learning-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java
@@ -0,0 +1,116 @@
+package cn.ac.iie.utils;
+
+import cn.ac.iie.config.ApplicationConfig;
+import com.arangodb.ArangoCollection;
+import com.arangodb.ArangoCursor;
+import com.arangodb.ArangoDB;
+import com.arangodb.ArangoDatabase;
+import com.arangodb.entity.DocumentCreateEntity;
+import com.arangodb.entity.ErrorEntity;
+import com.arangodb.entity.MultiDocumentEntity;
+import com.arangodb.model.AqlQueryOptions;
+import com.arangodb.model.DocumentCreateOptions;
+import com.arangodb.util.MapBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+public class ArangoDBConnect {
+ private static final Logger LOG = LoggerFactory.getLogger(ArangoDBConnect.class);
+ private static ArangoDB arangoDB = null;
+ private static ArangoDBConnect conn = null;
+ static {
+ getArangoDatabase();
+ }
+
+ private static void getArangoDatabase(){
+ arangoDB = new ArangoDB.Builder()
+ .maxConnections(ApplicationConfig.THREAD_POOL_NUMBER())
+ .host(ApplicationConfig.ARANGODB_HOST(), ApplicationConfig.ARANGODB_PORT())
+ .user(ApplicationConfig.ARANGODB_USER())
+ .password(ApplicationConfig.ARANGODB_PASSWORD())
+ .build();
+ }
+
+ public static synchronized ArangoDBConnect getInstance(){
+ if (null == conn){
+ conn = new ArangoDBConnect();
+ }
+ return conn;
+ }
+
+ private ArangoDatabase getDatabase(){
+ return arangoDB.db(ApplicationConfig.ARANGODB_DB_NAME());
+ }
+
+ public void clean(){
+ try {
+ if (arangoDB != null){
+ arangoDB.shutdown();
+ }
+ }catch (Exception e){
+ e.printStackTrace();
+ }
+ }
+
+ public ArangoCursor executorQuery(String query,Class type){
+ ArangoDatabase database = getDatabase();
+ Map bindVars = new MapBuilder().get();
+ AqlQueryOptions options = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL());
+ try {
+ return database.query(query, bindVars, options, type);
+ }catch (Exception e){
+ e.printStackTrace();
+ return null;
+ }finally {
+ bindVars.clear();
+ }
+ }
+
+ @Deprecated
+ public void insertAndUpdate(ArrayList docInsert,ArrayList docUpdate,String collectionName){
+ ArangoDatabase database = getDatabase();
+ try {
+ ArangoCollection collection = database.collection(collectionName);
+ if (!docInsert.isEmpty()){
+ collection.importDocuments(docInsert);
+ }
+ if (!docUpdate.isEmpty()){
+ collection.replaceDocuments(docUpdate);
+ }
+ }catch (Exception e){
+ System.out.println("更新失败");
+ e.printStackTrace();
+ }finally {
+ docInsert.clear();
+ docInsert.clear();
+ }
+ }
+
+ public void overwrite(ArrayList docOverwrite,String collectionName){
+ ArangoDatabase database = getDatabase();
+ try {
+ ArangoCollection collection = database.collection(collectionName);
+ if (!docOverwrite.isEmpty()){
+ DocumentCreateOptions documentCreateOptions = new DocumentCreateOptions();
+ documentCreateOptions.overwrite(true);
+ documentCreateOptions.silent(true);
+ MultiDocumentEntity> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
+ Collection errors = documentCreateEntityMultiDocumentEntity.getErrors();
+ for (ErrorEntity errorEntity:errors){
+ LOG.warn("写入arangoDB异常："+errorEntity.getErrorMessage());
+ }
+ }
+ }catch (Exception e){
+ System.out.println("更新失败："+e.toString());
+ }finally {
+ docOverwrite.clear();
+ }
+ }
+
+
+
+}
diff --git a/ip-learning-spark/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java b/ip-learning-spark/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
new file mode 100644
index 0000000..b9190d3
--- /dev/null
+++ b/ip-learning-spark/src/main/java/cn/ac/iie/utils/ExecutorThreadPool.java
@@ -0,0 +1,48 @@
+package cn.ac.iie.utils;
+
+import cn.ac.iie.config.ApplicationConfig;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.*;
+
+/**
+ * 线程池管理
+ * @author wlh
+ */
+public class ExecutorThreadPool {
+ private static ExecutorService pool = null ;
+ private static ExecutorThreadPool poolExecutor = null;
+
+ static {
+ getThreadPool();
+ }
+
+ private static void getThreadPool(){
+ ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("iplearning-application-pool-%d").build();
+
+ //Common Thread Pool
+ pool = new ThreadPoolExecutor(ApplicationConfig.THREAD_POOL_NUMBER(), ApplicationConfig.THREAD_POOL_NUMBER()*2,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
+
+// pool = Executors.newFixedThreadPool(ApplicationConfig.THREAD_POOL_NUMBER);
+ }
+
+ public static ExecutorThreadPool getInstance(){
+ if (null == poolExecutor){
+ poolExecutor = new ExecutorThreadPool();
+ }
+ return poolExecutor;
+ }
+
+ public void executor(Runnable command){
+ pool.execute(command);
+ }
+
+ public void shutdown(){
+ pool.shutdown();
+ }
+
+
+}
diff --git a/ip-learning-spark/src/main/resources/application.properties b/ip-learning-spark/src/main/resources/application.properties
new file mode 100644
index 0000000..473771f
--- /dev/null
+++ b/ip-learning-spark/src/main/resources/application.properties
@@ -0,0 +1,45 @@
+#spark任务配置
+spark.sql.shuffle.partitions=5
+spark.executor.memory=4g
+spark.app.name=test
+spark.network.timeout=300s
+repartitionNumber=36
+spark.serializer=org.apache.spark.serializer.KryoSerializer
+master=local[*]
+#spark读取clickhouse配置
+spark.read.clickhouse.url=jdbc:clickhouse://192.168.40.186:8123/tsg_galaxy_v3
+spark.read.clickhouse.driver=ru.yandex.clickhouse.ClickHouseDriver
+spark.read.clickhouse.user=default
+spark.read.clickhouse.password=111111
+spark.read.clickhouse.numPartitions=144
+spark.read.clickhouse.fetchsize=10000
+spark.read.clickhouse.partitionColumn=common_recv_time
+clickhouse.socket.timeout=300000
+#arangoDB配置
+arangoDB.host=192.168.40.182
+arangoDB.port=8529
+arangoDB.user=upsert
+arangoDB.password=ceiec2018
+#arangoDB.DB.name=insert_iplearn_index
+arangoDB.DB.name=ip-learning-test-0
+arangoDB.ttl=3600
+
+thread.pool.number=5
+
+#读取clickhouse时间范围方式，0：读取过去一小时；1：指定时间范围
+clickhouse.time.limit.type=0
+read.clickhouse.max.time=1571245220
+read.clickhouse.min.time=1571245210
+
+#读取arangoDB时间范围方式，0：正常读；1：指定时间范围
+arango.time.limit.type=0
+read.arango.max.time=1571245320
+read.arango.min.time=1571245200
+
+arangoDB.read.limit=
+update.arango.batch=10000
+
+distinct.client.ip.num=10000
+recent.count.hour=24
+
+update.interval=3600
diff --git a/ip-learning-spark/src/main/resources/log4j.properties b/ip-learning-spark/src/main/resources/log4j.properties
new file mode 100644
index 0000000..5d836b5
--- /dev/null
+++ b/ip-learning-spark/src/main/resources/log4j.properties
@@ -0,0 +1,25 @@
+######################### logger ##############################
+log4j.logger.org.apache.http=OFF
+log4j.logger.org.apache.http.wire=OFF
+
+#Log4j
+log4j.rootLogger=info,console,file
+# ����̨��־����
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Threshold=warn
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] [Thread\:%t] %l %x - <%m>%n
+
+# �ļ���־����
+log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.file.Threshold=warn
+log4j.appender.file.encoding=UTF-8
+log4j.appender.file.Append=true
+#·���������·����������ز��������Ӧ��Ŀ��
+#log4j.appender.file.file=/home/ceiec/iplearning/logs/ip-learning-application.log
+#log4j.appender.file.file=/home/ceiec/iplearning/testLog/ip-learning-application.log
+log4j.appender.file.file=./logs/ip-learning-application.log
+log4j.appender.file.DatePattern='.'yyyy-MM-dd
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+#log4j.appender.file.layout.ConversionPattern=%d{HH:mm:ss} %X{ip} [%t] %5p %c{1} %m%n
+log4j.appender.file.layout.ConversionPattern=[%d{yyyy-MM-dd HH\:mm\:ss}] [%-5p] %X{ip} [Thread\:%t] %l %x - %m%n
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
new file mode 100644
index 0000000..395ea6b
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/config/ApplicationConfig.scala
@@ -0,0 +1,51 @@
+package cn.ac.iie.config
+
+import com.typesafe.config.{Config, ConfigFactory}
+
+object ApplicationConfig {
+ private lazy val config: Config = ConfigFactory.load()
+
+ val SPARK_SQL_SHUFFLE_PARTITIONS: Int = config.getInt("spark.sql.shuffle.partitions")
+ val SPARK_EXECUTOR_MEMORY: String = config.getString("spark.executor.memory")
+ 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")
+ val MASTER: String = config.getString("master")
+ val SPARK_SERIALIZER: String = config.getString("spark.serializer")
+
+ val NUMPARTITIONS: String = config.getString("spark.read.clickhouse.numPartitions")
+ val SPARK_READ_CLICKHOUSE_URL: String = config.getString("spark.read.clickhouse.url")
+ val SPARK_READ_CLICKHOUSE_DRIVER: String = config.getString("spark.read.clickhouse.driver")
+ val SPARK_READ_CLICKHOUSE_USER: String = config.getString("spark.read.clickhouse.user")
+ val SPARK_READ_CLICKHOUSE_PASSWORD: String = config.getString("spark.read.clickhouse.password")
+ val SPARK_READ_CLICKHOUSE_FETCHSIZE: String = config.getString("spark.read.clickhouse.fetchsize")
+ val SPARK_READ_CLICKHOUSE_PARTITIONCOLUMN: String = config.getString("spark.read.clickhouse.partitionColumn")
+
+ val ARANGODB_HOST: String= config.getString("arangoDB.host")
+ val ARANGODB_PORT: Int = config.getInt("arangoDB.port")
+ val ARANGODB_USER: String= config.getString("arangoDB.user")
+ val ARANGODB_PASSWORD:String= config.getString("arangoDB.password")
+ val ARANGODB_DB_NAME:String= config.getString("arangoDB.DB.name")
+ val ARANGODB_TTL: Int = config.getInt("arangoDB.ttl")
+ val CLICKHOUSE_SOCKET_TIMEOUT: Int = config.getInt("clickhouse.socket.timeout")
+
+ val THREAD_POOL_NUMBER: Int = config.getInt("thread.pool.number")
+
+ val CLICKHOUSE_TIME_LIMIT_TYPE: Int = config.getInt("clickhouse.time.limit.type")
+
+ val READ_CLICKHOUSE_MAX_TIME: Long = config.getLong("read.clickhouse.max.time")
+ val READ_CLICKHOUSE_MIN_TIME: Long = config.getLong("read.clickhouse.min.time")
+
+ val ARANGO_TIME_LIMIT_TYPE: Int = config.getInt("arango.time.limit.type")
+
+ val READ_ARANGO_MAX_TIME: Long = config.getLong("read.arango.max.time")
+ val READ_ARANGO_MIN_TIME: Long = config.getLong("read.arango.min.time")
+
+ val ARANGODB_READ_LIMIT: String = config.getString("arangoDB.read.limit")
+ val UPDATE_ARANGO_BATCH: Int = config.getInt("update.arango.batch")
+ val RECENT_COUNT_HOUR: Int = config.getInt("recent.count.hour")
+ val DISTINCT_CLIENT_IP_NUM: Int = config.getInt("distinct.client.ip.num")
+
+ val UPDATE_INTERVAL: Int = config.getInt("update.interval")
+
+}
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
new file mode 100644
index 0000000..952c30c
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/dao/BaseClickhouseData.scala
@@ -0,0 +1,209 @@
+package cn.ac.iie.dao
+
+import cn.ac.iie.config.ApplicationConfig
+import cn.ac.iie.utils.SparkSessionUtil.spark
+import org.apache.spark.sql.DataFrame
+import org.slf4j.LoggerFactory
+
+object BaseClickhouseData {
+ private val LOG = LoggerFactory.getLogger(BaseClickhouseData.getClass)
+
+ val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60
+ private val timeLimit: (Long, Long) = getTimeLimit
+
+ private def initClickhouseData(sql:String): Unit ={
+
+ val dataFrame: DataFrame = spark.read.format("jdbc")
+ .option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL)
+ .option("dbtable", sql)
+ .option("driver", ApplicationConfig.SPARK_READ_CLICKHOUSE_DRIVER)
+ .option("user", ApplicationConfig.SPARK_READ_CLICKHOUSE_USER)
+ .option("password", ApplicationConfig.SPARK_READ_CLICKHOUSE_PASSWORD)
+ .option("numPartitions", ApplicationConfig.NUMPARTITIONS)
+ .option("partitionColumn", ApplicationConfig.SPARK_READ_CLICKHOUSE_PARTITIONCOLUMN)
+ .option("lowerBound", timeLimit._2)
+ .option("upperBound", timeLimit._1)
+ .option("fetchsize", ApplicationConfig.SPARK_READ_CLICKHOUSE_FETCHSIZE)
+ .option("socket_timeout",ApplicationConfig.CLICKHOUSE_SOCKET_TIMEOUT)
+ .load()
+ dataFrame.printSchema()
+ dataFrame.createOrReplaceGlobalTempView("dbtable")
+ }
+
+ def loadConnectionDataFromCk(): Unit ={
+ val where = "common_recv_time >= " + timeLimit._2 + " AND common_recv_time < " + timeLimit._1
+ val sql =
+ s"""
+ |(SELECT
+ | ssl_sni,http_host,common_client_ip,common_server_ip,common_recv_time,common_c2s_byte_num,common_s2c_byte_num,common_schema_type
+ |FROM
+ | connection_record_log
+ |WHERE $where) as dbtable
+ """.stripMargin
+
+ LOG.warn(sql)
+ initClickhouseData(sql)
+ }
+
+ private def loadRadiusDataFromCk(): Unit ={
+ val where =
+ s"""
+ | common_recv_time >= ${timeLimit._2}
+ | AND common_recv_time < ${timeLimit._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,common_recv_time
+ |FROM
+ | tsg_galaxy_v3.radius_record_log
+ |WHERE
+ | $where) as dbtable
+ """.stripMargin
+ LOG.warn(sql)
+ initClickhouseData(sql)
+ }
+
+ def getVertexFqdnDf: DataFrame ={
+ loadConnectionDataFromCk()
+ val sql =
+ """
+ |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
+ | global_temp.dbtable
+ | WHERE
+ | 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
+ | global_temp.dbtable
+ | WHERE
+ | common_schema_type = 'HTTP' GROUP BY http_host
+ | )
+ | )
+ |GROUP BY
+ | FQDN
+ |HAVING
+ | FQDN != ''
+ """.stripMargin
+ LOG.warn(sql)
+ val vertexFqdnDf = spark.sql(sql)
+ vertexFqdnDf.printSchema()
+ vertexFqdnDf
+ }
+
+ def getVertexIpDf: DataFrame ={
+ loadConnectionDataFromCk()
+ val sql =
+ """
+ |SELECT
+ | *
+ |FROM
+ | (
+ | (
+ | 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) as BYTES_SUM,
+ | 'client' as ip_type
+ | FROM
+ | global_temp.dbtable
+ | GROUP BY
+ | IP
+ | )
+ | UNION ALL
+ | (
+ | 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_s2c_byte_num) as BYTES_SUM,
+ | 'server' as ip_type
+ | FROM
+ | global_temp.dbtable
+ | GROUP BY
+ | IP
+ | )
+ | )
+ """.stripMargin
+ LOG.warn(sql)
+ val vertexIpDf = spark.sql(sql)
+ vertexIpDf.printSchema()
+ vertexIpDf
+ }
+
+ def getRelationFqdnLocateIpDf: DataFrame ={
+ loadConnectionDataFromCk()
+ val 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,
+ | collect_set(common_client_ip) AS DIST_CIP_RECENT,
+ | 'TLS' AS schema_type
+ |FROM
+ | global_temp.dbtable
+ |WHERE
+ | common_schema_type = 'SSL'
+ |GROUP BY
+ | ssl_sni,common_server_ip
+ """.stripMargin
+
+ val 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,
+ | collect_set(common_client_ip) AS DIST_CIP_RECENT,
+ | 'HTTP' AS schema_type
+ |FROM
+ | global_temp.dbtable
+ |WHERE
+ | common_schema_type = 'HTTP'
+ |GROUP BY
+ | http_host,common_server_ip
+ """.stripMargin
+ val sql = s"SELECT * FROM (($sslSql) UNION ALL ($httpSql)) WHERE FQDN != ''"
+
+ LOG.warn(sql)
+ val relationFqdnLocateIpDf = spark.sql(sql)
+ relationFqdnLocateIpDf.printSchema()
+ relationFqdnLocateIpDf
+ }
+
+ private def getTimeLimit: (Long,Long) ={
+ var maxTime = 0L
+ var minTime = 0L
+ ApplicationConfig.CLICKHOUSE_TIME_LIMIT_TYPE match {
+ case 0 =>
+ maxTime = currentHour
+ minTime = maxTime - ApplicationConfig.UPDATE_INTERVAL
+ case 1 =>
+ maxTime = ApplicationConfig.READ_CLICKHOUSE_MAX_TIME
+ minTime = ApplicationConfig.READ_CLICKHOUSE_MIN_TIME
+ case _ =>
+ }
+ (maxTime, minTime)
+ }
+
+}
diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpLearningApplication.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpLearningApplication.scala
new file mode 100644
index 0000000..e3602d3
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/main/IpLearningApplication.scala
@@ -0,0 +1,22 @@
+package cn.ac.iie.main
+
+import cn.ac.iie.service.update.UpdateDocument._
+import cn.ac.iie.utils.{ExecutorThreadPool, SparkSessionUtil}
+
+object IpLearningApplication {
+ private val pool = ExecutorThreadPool.getInstance
+
+ def main(args: Array[String]): Unit = {
+ try {
+ updateVertexFqdn()
+ updateVertexIp()
+ updateRelationFqdnLocateIp()
+ }catch {
+ case e:Exception => e.printStackTrace()
+ }finally {
+ pool.shutdown()
+ arangoManger.clean()
+ SparkSessionUtil.closeSpark()
+ }
+ }
+}
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
new file mode 100644
index 0000000..460caed
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/service/transform/MergeDataFrame.scala
@@ -0,0 +1,88 @@
+package cn.ac.iie.service.transform
+
+import java.util.regex.Pattern
+
+import cn.ac.iie.config.ApplicationConfig
+import cn.ac.iie.dao.BaseClickhouseData
+import cn.ac.iie.spark.partition.CustomPartitioner
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions._
+import org.slf4j.LoggerFactory
+
+object MergeDataFrame {
+ private val LOG = LoggerFactory.getLogger(MergeDataFrame.getClass)
+ private val pattern = Pattern.compile("^[\\d]*$")
+
+ def mergeVertexFqdn(): RDD[Row] ={
+ BaseClickhouseData.getVertexFqdnDf
+ .rdd.filter(row => isDomain(row.getAs[String](0))).map(row => (row.get(0),row))
+ .partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
+ }
+
+ def mergeVertexIp(): RDD[Row]={
+ val vertexIpDf = BaseClickhouseData.getVertexIpDf
+ val frame = vertexIpDf.groupBy("IP").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")
+ )
+ val values = frame.rdd.map(row => (row.get(0), row))
+ .partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
+ values
+ }
+
+ def mergeRelationFqdnLocateIp(): RDD[Row] ={
+ val frame = BaseClickhouseData.getRelationFqdnLocateIpDf.filter(row => isDomain(row.getAs[String]("FQDN")))
+ .groupBy("FQDN", "common_server_ip")
+ .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")
+ )
+ frame.rdd.map(row => {
+ val fqdn = row.getAs[String]("FQDN")
+ val serverIp = row.getAs[String]("common_server_ip")
+ val key = fqdn.concat("-"+serverIp)
+ (key,row)
+ }).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
+
+ }
+
+ private def isDomain(fqdn: String): Boolean = {
+ try {
+ if (fqdn == null || fqdn.length == 0) {
+ return false
+ }
+ if (fqdn.contains(":")) {
+ val s = fqdn.split(":")(0)
+ if (s.contains(":")){
+ return false
+ }
+ }
+ val fqdnArr = fqdn.split("\\.")
+ if (fqdnArr.length < 4 || fqdnArr.length > 4){
+ return true
+ }
+ for (f <- fqdnArr) {
+ if (pattern.matcher(f).matches) {
+ val i = f.toLong
+ if (i < 0 || i > 255) {
+ return true
+ }
+ } else {
+ return true
+ }
+ }
+ } catch {
+ case e: Exception =>
+ LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString)
+ }
+ false
+ }
+
+}
diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocHandler.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocHandler.scala
new file mode 100644
index 0000000..bdf8120
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocHandler.scala
@@ -0,0 +1,123 @@
+package cn.ac.iie.service.update
+
+
+import java.lang
+
+import cn.ac.iie.config.ApplicationConfig
+import cn.ac.iie.service.read.ReadHistoryArangoData
+import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
+
+import scala.collection.mutable
+import scala.collection.mutable.WrappedArray.ofRef
+
+object UpdateDocHandler {
+ val PROTOCOL_SET: Set[String] = Set("HTTP","TLS","DNS")
+
+ def updateMaxAttribute(hisDoc: BaseDocument,newAttribute:Long,attributeName:String): Unit ={
+ var hisAttritube = hisDoc.getAttribute(attributeName).toString.toLong
+ if (newAttribute > hisAttritube){
+ hisAttritube = newAttribute
+ }
+ hisDoc.addAttribute(attributeName,hisAttritube)
+ }
+
+ def updateSumAttribute(hisDoc: BaseDocument,newAttribute:Long,attributeName:String): Unit ={
+ val hisAttritube = hisDoc.getAttribute(attributeName).toString.toLong
+ hisDoc.addAttribute(attributeName,newAttribute+hisAttritube)
+ }
+
+ def separateAttributeByIpType(ipTypeList:ofRef[String],
+ sessionCountList:ofRef[AnyRef],
+ bytesSumList:ofRef[AnyRef]): (Long,Long,Long,Long) ={
+ var serverSessionCount = 0L
+ var serverBytesSum = 0L
+ var clientSessionCount = 0L
+ var clientBytesSum = 0L
+ if (ipTypeList.length == sessionCountList.length && ipTypeList.length == bytesSumList.length){
+ sessionCountList.zip(bytesSumList).zip(ipTypeList).foreach(t => {
+ t._2 match {
+ case "server" =>
+ serverSessionCount = t._1._1.toString.toLong
+ serverBytesSum = t._1._2.toString.toLong
+ case "client" =>
+ clientSessionCount = t._1._1.toString.toLong
+ clientBytesSum = t._1._2.toString.toLong
+ }
+ })
+ }
+ (serverSessionCount, serverBytesSum, clientSessionCount, clientBytesSum)
+ }
+
+ def separateAttributeByProtocol(schemaTypeList:ofRef[AnyRef],countTotalList:ofRef[AnyRef]): Map[String, Long] ={
+ var protocolMap: Map[String, Long] = Map()
+ if (schemaTypeList.length == countTotalList.length){
+ protocolMap = schemaTypeList.zip(countTotalList).map(t => (t._1.toString,t._2.toString.toLong)).toMap
+ }
+ PROTOCOL_SET.foreach(protocol => {
+ if (!protocolMap.contains(protocol)){
+ protocolMap += (protocol -> 0L)
+ }
+ })
+ protocolMap
+ }
+
+ def updateProtocolAttritube(hisDoc:BaseEdgeDocument, protocolMap: Map[String, Long]): Unit ={
+ var protocolType = hisDoc.getAttribute("PROTOCOL_TYPE").toString
+ protocolMap.foreach(t => {
+ if (t._2 > 0 && !protocolType.contains(t._1)){
+ protocolType = protocolType.concat(","+ t._1)
+ }
+ val cntTotalName = t._1.concat("_CNT_TOTAL")
+ val cntRecentName = t._1.concat("_CNT_RECENT")
+ val cntRecent: Array[lang.Long] = hisDoc.getAttribute(cntRecentName).asInstanceOf[Array[java.lang.Long]]
+ cntRecent.update(0,t._2)
+ updateSumAttribute(hisDoc,t._2,cntTotalName)
+ hisDoc.addAttribute(cntRecentName,cntRecent)
+ })
+ hisDoc.addAttribute("PROTOCOL_TYPE",protocolType)
+ }
+
+ def putProtocolAttritube(doc:BaseEdgeDocument, protocolMap: Map[String, Long]): Unit ={
+ val protocolTypeBuilder = new mutable.StringBuilder()
+ protocolMap.foreach(t => {
+ if (t._2 > 0){
+ protocolTypeBuilder.append(","+t._1)
+ }
+ val cntTotalName = t._1.concat("_CNT_TOTAL")
+ val cntRecentName = t._1.concat("_CNT_RECENT")
+ val cntRecent: Array[Long] = new Array[Long](ApplicationConfig.RECENT_COUNT_HOUR)
+ cntRecent.update(0,t._2)
+ doc.addAttribute(cntTotalName,t._2)
+ doc.addAttribute(cntRecentName,cntRecent)
+ })
+ doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",",""))
+ }
+
+ def mergeDistinctIp(distCipRecent:ofRef[ofRef[String]]): Array[String] ={
+ distCipRecent.flatten.distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
+ }
+
+ def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={
+ val map = newDistinctIp.map(ip => {
+ (ip, ReadHistoryArangoData.currentHour)
+ }).toMap
+ doc.addAttribute("DIST_CIP",map.keys.toArray)
+ doc.addAttribute("DIST_CIP_TS",map.values.toArray)
+ }
+
+ def updateDistinctIp(hisDoc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={
+ val hisDistCip = hisDoc.getAttribute("DIST_CIP").asInstanceOf[Array[String]]
+ val hisDistCipTs = hisDoc.getAttribute("DIST_CIP_TS").asInstanceOf[Array[Long]]
+ if (hisDistCip.length == hisDistCipTs.length){
+ val distCipToTsMap: Map[String, Long] = hisDistCip.zip(hisDistCipTs).toMap
+ val muDistCipToTsMap: mutable.Map[String, Long] = mutable.Map(distCipToTsMap.toSeq:_*)
+ newDistinctIp.foreach(cip => {
+ muDistCipToTsMap.put(cip,ReadHistoryArangoData.currentHour)
+ })
+ val resultMap = muDistCipToTsMap.toList.sortBy(-_._2).take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toMap
+ hisDoc.addAttribute("DIST_CIP",resultMap.keys.toArray)
+ hisDoc.addAttribute("DIST_CIP_TS",resultMap.values.toArray)
+ }
+ }
+
+}
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
new file mode 100644
index 0000000..b5f875f
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/service/update/UpdateDocument.scala
@@ -0,0 +1,194 @@
+package cn.ac.iie.service.update
+import java.util
+
+import cn.ac.iie.config.ApplicationConfig
+import cn.ac.iie.dao.BaseArangoData
+import cn.ac.iie.dao.BaseArangoData._
+import cn.ac.iie.service.transform.MergeDataFrame._
+import cn.ac.iie.service.update.UpdateDocHandler._
+import cn.ac.iie.utils.ArangoDBConnect
+import cn.ac.iie.utils.SparkSessionUtil.spark
+import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
+import org.apache.spark.TaskContext
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Row
+import org.slf4j.LoggerFactory
+
+import scala.collection.mutable.WrappedArray.ofRef
+
+object UpdateDocument {
+
+ val arangoManger: ArangoDBConnect = ArangoDBConnect.getInstance()
+ private val LOG = LoggerFactory.getLogger(UpdateDocument.getClass)
+ private val baseArangoData = new BaseArangoData()
+
+ def updateVertexFqdn(): Unit ={
+ baseArangoData.readHistoryData("FQDN",historyVertexFqdnMap,classOf[BaseDocument])
+ val hisVerFqdnBc = spark.sparkContext.broadcast(historyVertexFqdnMap)
+ try {
+ val start = System.currentTimeMillis()
+ val mergeVertexFqdnDf: RDD[Row] = mergeVertexFqdn()
+ mergeVertexFqdnDf.foreachPartition(iter => {
+ val partitionId: Int = TaskContext.get.partitionId
+ val hisVerFqdnMapTmp = hisVerFqdnBc.value.get(partitionId)
+ val resultDocumentList: util.ArrayList[BaseDocument] = new util.ArrayList[BaseDocument]
+ var i = 0
+ iter.foreach(row => {
+ val fqdn = row.getAs[String]("FQDN")
+ val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
+ val firstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
+ var document: BaseDocument = hisVerFqdnMapTmp.getOrDefault(fqdn,null)
+ if (document != null){
+ updateMaxAttribute(document,lastFoundTime,"LAST_FOUND_TIME")
+ } else{
+ document = new BaseDocument
+ document.setKey(fqdn)
+ document.addAttribute("FQDN_NAME",fqdn)
+ document.addAttribute("FIRST_FOUND_TIME",firstFoundTime)
+ document.addAttribute("LAST_FOUND_TIME",lastFoundTime)
+ }
+ resultDocumentList.add(document)
+ i+=1
+ if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH){
+ arangoManger.overwrite(resultDocumentList, "FQDN")
+ LOG.warn("更新FQDN:" + i)
+ i = 0
+ }
+ })
+ if (i != 0) {
+ arangoManger.overwrite(resultDocumentList, "FQDN")
+ LOG.warn("更新FQDN:" + i)
+ }
+ })
+ val last = System.currentTimeMillis()
+ LOG.warn(s"更新FQDN时间：${last-start}")
+ }catch {
+ case e:Exception => e.printStackTrace()
+ }finally {
+ hisVerFqdnBc.destroy()
+ }
+ }
+
+ def updateVertexIp(): Unit ={
+ baseArangoData.readHistoryData("IP",historyVertexIpMap,classOf[BaseDocument])
+ val hisVerIpBc = spark.sparkContext.broadcast(historyVertexIpMap)
+ try {
+ val start = System.currentTimeMillis()
+ val mergeVertexIpDf = mergeVertexIp()
+ mergeVertexIpDf.foreachPartition(iter => {
+ val partitionId: Int = TaskContext.get.partitionId
+ val hisVerIpMapTmp = hisVerIpBc.value.get(partitionId)
+ val resultDocumentList: util.ArrayList[BaseDocument] = new util.ArrayList[BaseDocument]
+ var i = 0
+ iter.foreach(row => {
+ val ip = row.getAs[String]("IP")
+ val firstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
+ val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
+ val sessionCountList = row.getAs[ofRef[AnyRef]]("SESSION_COUNT_LIST")
+ val bytesSumList = row.getAs[ofRef[AnyRef]]("BYTES_SUM_LIST")
+ val ipTypeList = row.getAs[ofRef[String]]("ip_type_list")
+ val sepAttributeTuple = separateAttributeByIpType(ipTypeList,sessionCountList,bytesSumList)
+
+ var document = hisVerIpMapTmp.getOrDefault(ip,null)
+ if (document != null){
+ updateMaxAttribute(document,lastFoundTime,"LAST_FOUND_TIME")
+ updateSumAttribute(document,sepAttributeTuple._1,"SERVER_SESSION_COUNT")
+ updateSumAttribute(document,sepAttributeTuple._2,"SERVER_BYTES_SUM")
+ updateSumAttribute(document,sepAttributeTuple._3,"CLIENT_SESSION_COUNT")
+ updateSumAttribute(document,sepAttributeTuple._4,"CLIENT_BYTES_SUM")
+ } else {
+ document = new BaseDocument
+ document.setKey(ip)
+ document.addAttribute("IP",ip)
+ document.addAttribute("FIRST_FOUND_TIME",firstFoundTime)
+ document.addAttribute("LAST_FOUND_TIME",lastFoundTime)
+ document.addAttribute("SERVER_SESSION_COUNT",sepAttributeTuple._1)
+ document.addAttribute("SERVER_BYTES_SUM",sepAttributeTuple._2)
+ document.addAttribute("CLIENT_SESSION_COUNT",sepAttributeTuple._3)
+ document.addAttribute("CLIENT_BYTES_SUM",sepAttributeTuple._4)
+ document.addAttribute("COMMON_LINK_INFO","")
+ }
+ resultDocumentList.add(document)
+ i+=1
+ if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH){
+ arangoManger.overwrite(resultDocumentList, "IP")
+ LOG.warn("更新IP:" + i)
+ i = 0
+ }
+ })
+ if (i != 0) {
+ arangoManger.overwrite(resultDocumentList, "IP")
+ LOG.warn("更新IP:" + i)
+ }
+ })
+ val last = System.currentTimeMillis()
+ LOG.warn(s"更新IP时间：${last-start}")
+ }catch {
+ case e:Exception => e.printStackTrace()
+ }finally {
+ hisVerIpBc.destroy()
+ }
+ }
+
+ def updateRelationFqdnLocateIp(): Unit ={
+ baseArangoData.readHistoryData("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, classOf[BaseEdgeDocument])
+ val hisReFqdnLocIpBc = spark.sparkContext.broadcast(historyRelationFqdnAddressIpMap)
+ try {
+ val start = System.currentTimeMillis()
+ val mergeRelationFqdnLocateIpDf = mergeRelationFqdnLocateIp()
+ mergeRelationFqdnLocateIpDf.foreachPartition(iter => {
+ val partitionId: Int = TaskContext.get.partitionId
+ val hisRelaFqdnLocaIpMapTmp = hisReFqdnLocIpBc.value.get(partitionId)
+ val resultDocumentList: util.ArrayList[BaseEdgeDocument] = new util.ArrayList[BaseEdgeDocument]
+ var i = 0
+ iter.foreach(row => {
+ val fqdn = row.getAs[String]("FQDN")
+ val serverIp = row.getAs[String]("common_server_ip")
+ val firstFoundTime = row.getAs[Long]("FIRST_FOUND_TIME")
+ val lastFoundTime = row.getAs[Long]("LAST_FOUND_TIME")
+ val countTotalList = row.getAs[ofRef[AnyRef]]("COUNT_TOTAL_LIST")
+ val schemaTypeList = row.getAs[ofRef[AnyRef]]("schema_type_list")
+ val distCipRecent = row.getAs[ofRef[ofRef[String]]]("DIST_CIP_RECENT")
+
+ val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList,countTotalList)
+ val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
+
+ val key = fqdn.concat("-"+serverIp)
+ var document: BaseEdgeDocument = hisRelaFqdnLocaIpMapTmp.getOrDefault(key,null)
+ if (document != null){
+ updateMaxAttribute(document,lastFoundTime,"LAST_FOUND_TIME")
+ updateProtocolAttritube(document,sepAttritubeMap)
+ updateDistinctIp(document,distinctIp)
+ }else {
+ document = new BaseEdgeDocument()
+ document.setKey(key)
+ document.setFrom("FQDN/" + fqdn)
+ document.setTo("IP/" + serverIp)
+ document.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
+ document.addAttribute("LAST_FOUND_TIME", lastFoundTime)
+ putProtocolAttritube(document,sepAttritubeMap)
+ putDistinctIp(document,distinctIp)
+ }
+ resultDocumentList.add(document)
+ i+=1
+ if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH){
+ arangoManger.overwrite(resultDocumentList, "R_LOCATE_FQDN2IP")
+ LOG.warn("更新R_LOCATE_FQDN2IP:" + i)
+ i = 0
+ }
+ })
+ if (i != 0) {
+ arangoManger.overwrite(resultDocumentList, "R_LOCATE_FQDN2IP")
+ LOG.warn("更新R_LOCATE_FQDN2IP:" + i)
+ }
+ })
+ val last = System.currentTimeMillis()
+ LOG.warn(s"更新R_LOCATE_FQDN2IP时间：${last-start}")
+ }catch {
+ case e:Exception => e.printStackTrace()
+ }finally {
+ hisReFqdnLocIpBc.destroy()
+ }
+ }
+
+}
diff --git a/ip-learning-spark/src/main/scala/cn/ac/iie/spark/partition/CustomPartitioner.scala b/ip-learning-spark/src/main/scala/cn/ac/iie/spark/partition/CustomPartitioner.scala
new file mode 100644
index 0000000..f33a43e
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/spark/partition/CustomPartitioner.scala
@@ -0,0 +1,11 @@
+package cn.ac.iie.spark.partition
+
+import org.apache.spark.Partitioner
+
+class CustomPartitioner(numPartition: Int) extends Partitioner{
+ override def numPartitions: Int = numPartition
+
+ override def getPartition(key: Any): Int = {
+ Math.abs(key.hashCode()) % numPartition
+ }
+}
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
new file mode 100644
index 0000000..12cfc86
--- /dev/null
+++ b/ip-learning-spark/src/main/scala/cn/ac/iie/utils/SparkSessionUtil.scala
@@ -0,0 +1,32 @@
+package cn.ac.iie.utils
+
+import cn.ac.iie.config.ApplicationConfig
+import org.apache.spark.sql.SparkSession
+import org.slf4j.LoggerFactory
+
+object SparkSessionUtil {
+ private val LOG = LoggerFactory.getLogger(SparkSessionUtil.getClass)
+
+ val spark: SparkSession = getSparkSession
+
+ private def getSparkSession: SparkSession ={
+ val spark: SparkSession = SparkSession
+ .builder()
+ .appName(ApplicationConfig.SPARK_APP_NAME)
+ .config("spark.serializer", ApplicationConfig.SPARK_SERIALIZER)
+ .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)
+ .master(ApplicationConfig.MASTER)
+ .getOrCreate()
+ LOG.warn("sparkession获取成功！！！")
+ spark
+ }
+
+ def closeSpark(): Unit ={
+ if (spark != null){
+ spark.stop()
+ }
+ }
+
+}
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
new file mode 100644
index 0000000..608fb2d
--- /dev/null
+++ b/ip-learning-spark/src/test/scala/cn/ac/iie/dao/BaseClickhouseDataTest.scala
@@ -0,0 +1,43 @@
+package cn.ac.iie.dao
+
+import cn.ac.iie.utils.SparkSessionUtil
+import org.apache.spark.sql.SparkSession
+
+
+object BaseClickhouseDataTest {
+ private val spark: SparkSession = SparkSessionUtil.spark
+ def main(args: Array[String]): Unit = {
+ BaseClickhouseData loadConnectionDataFromCk()
+ val sql =
+ """
+ |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
+ | global_temp.dbtable
+ | WHERE
+ | 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
+ | global_temp.dbtable
+ | WHERE
+ | common_schema_type = 'HTTP' GROUP BY http_host
+ | )
+ | )
+ |GROUP BY
+ | FQDN
+ |HAVING
+ | FQDN != ''
+ """.stripMargin
+ println(sql)
+ val vertexFqdnDf = spark.sql(sql)
+ vertexFqdnDf.show(10)
+ }
+
+}
diff --git a/ip-learning-spark/src/test/scala/cn/ac/iie/service/update/UpdateDocumentTest.scala b/ip-learning-spark/src/test/scala/cn/ac/iie/service/update/UpdateDocumentTest.scala
new file mode 100644
index 0000000..67590ff
--- /dev/null
+++ b/ip-learning-spark/src/test/scala/cn/ac/iie/service/update/UpdateDocumentTest.scala
@@ -0,0 +1,35 @@
+package cn.ac.iie.service.update
+
+import java.util
+import java.util.ArrayList
+import java.util.concurrent.ConcurrentHashMap
+
+import cn.ac.iie.dao.BaseArangoData
+import cn.ac.iie.dao.BaseArangoData._
+import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
+
+import scala.collection.mutable.WrappedArray.ofRef
+
+object UpdateDocumentTest {
+ def main(args: Array[String]): Unit = {
+ val baseArangoData = new BaseArangoData()
+ baseArangoData.readHistoryData("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, classOf[BaseEdgeDocument])
+
+ val value = BaseArangoData.historyRelationFqdnAddressIpMap.keys()
+ while (value.hasMoreElements) {
+ val integer: Integer = value.nextElement()
+ val map: ConcurrentHashMap[String, BaseEdgeDocument] = historyRelationFqdnAddressIpMap.get(integer)
+ val unit = map.keys()
+ while (unit.hasMoreElements) {
+ val key = unit.nextElement()
+ val edgeDocument = map.get(key)
+ // val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[util.ArrayList[Long]]
+ // val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[util.ArrayList[String]]
+ val strings = edgeDocument.getAttribute("DIST_CIP").asInstanceOf[Array[String]]
+ val longs = edgeDocument.getAttribute("DNS_CNT_RECENT").asInstanceOf[Array[java.lang.Long]]
+ println(longs.toString + "---" + strings.toString)
+ }
+ }
+ }
+
+}