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

增加SUBSCRIBER点与R_LOCATE_SUBSCRIBER2IP关系

Browse Source
This commit is contained in:
wanglihui
2020-07-09 17:41:43 +08:00
parent 0e926aa7d0
commit d214e86a41
15 changed files with 263 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseArangoData.java
View File

@@ -1,7 +1,7 @@
package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.etl.read.ReadHistoryArangoData;
import cn.ac.iie.service.read.ReadHistoryArangoData;
import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.ArangoCursor;
@@ -15,10 +15,12 @@ import java.util.concurrent.ConcurrentHashMap;
public class BaseArangoData {
private static final Logger LOG = LoggerFactory.getLogger(BaseArangoData.class);
public static ConcurrentHashMap<String, BaseEdgeDocument> v_Fqdn_Map = new ConcurrentHashMap<>();
public static ConcurrentHashMap<String, BaseEdgeDocument> v_Ip_Map = new ConcurrentHashMap<>();
public static ConcurrentHashMap<String, BaseEdgeDocument> e_Fqdn_Address_Ip_Map = new ConcurrentHashMap<>();
public static ConcurrentHashMap<String, BaseEdgeDocument> e_Ip_Visit_Fqdn_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> v_Fqdn_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> v_Ip_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> v_Subscriber_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> e_Fqdn_Address_Ip_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> e_Ip_Visit_Fqdn_Map = new ConcurrentHashMap<>();
static ConcurrentHashMap<String, BaseEdgeDocument> e_Subsciber_Locate_Ip_Map = new ConcurrentHashMap<>();
private static final ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
@@ -28,14 +30,18 @@ public class BaseArangoData {
long startA = System.currentTimeMillis();
readHistoryData("FQDN", v_Fqdn_Map);
readHistoryData("IP", v_Ip_Map);
readHistoryData("SUBSCRIBER",v_Subscriber_Map);
readHistoryData("R_LOCATE_FQDN2IP", e_Fqdn_Address_Ip_Map);
readHistoryData("R_VISIT_IP2FQDN", e_Ip_Visit_Fqdn_Map);
readHistoryData("R_LOCATE_SUBSCRIBER2IP",e_Subsciber_Locate_Ip_Map);
threadPool.shutdown();
threadPool.awaitThreadTask();
LOG.info("v_Fqdn_Map大小："+BaseArangoData.v_Fqdn_Map.size());
LOG.info("v_Ip_Map大小："+BaseArangoData.v_Ip_Map.size());
LOG.info("e_Fqdn_Address_Ip_Map大小："+BaseArangoData.e_Fqdn_Address_Ip_Map.size());
LOG.info("e_Ip_Visit_Fqdn_Map大小："+BaseArangoData.e_Ip_Visit_Fqdn_Map.size());
LOG.info("v_Fqdn_Map大小："+v_Fqdn_Map.size());
LOG.info("v_Ip_Map大小："+v_Ip_Map.size());
LOG.info("v_Subscriber_Map大小："+v_Subscriber_Map.size());
LOG.info("e_Fqdn_Address_Ip_Map大小："+e_Fqdn_Address_Ip_Map.size());
LOG.info("e_Ip_Visit_Fqdn_Map大小："+e_Ip_Visit_Fqdn_Map.size());
LOG.info("e_Subsciber_Locate_Ip_Map大小："+e_Subsciber_Locate_Ip_Map.size());
long lastA = System.currentTimeMillis();
LOG.info("读取ArangoDb时间："+(lastA - startA));
}

81
IP-learning-graph/src/main/java/cn/ac/iie/dao/BaseClickhouseData.java
View File

@@ -14,23 +14,28 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import static cn.ac.iie.etl.read.ReadClickhouseData.*;
import static cn.ac.iie.service.read.ReadClickhouseData.*;
/**
* 读取clickhouse数据，封装到map
*/
public class BaseClickhouseData {
private static final Logger LOG = LoggerFactory.getLogger(BaseClickhouseData.class);
private static final ClickhouseConnect manger = ClickhouseConnect.getInstance();
static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> vFqdnMap = new HashMap<>();
static HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> vIpMap = new HashMap<>();
static HashMap<Integer,HashMap<String,ArrayList<BaseDocument>>> vSubscriberMap = new HashMap<>();
static HashMap<Integer, HashMap<String, HashMap<String, BaseEdgeDocument>>> eFqdnAddressIpMap = new HashMap<>();
static HashMap<Integer, HashMap<String, HashMap<String, BaseEdgeDocument>>> eIpVisitFqdnMap = new HashMap<>();
static HashMap<Integer, HashMap<String, HashMap<String, BaseEdgeDocument>>> eSubsciberLocateIpMap = new HashMap<>();
private DruidPooledConnection connection;
private Statement statement;
public void BaseVFqdn() {
void BaseVFqdn() {
initializeVertexMap(vFqdnMap);
LOG.info("V_FQDN resultMap初始化完成");
LOG.info("FQDN resultMap初始化完成");
String sql = getVFqdnSql();
long start = System.currentTimeMillis();
try {
@@ -57,9 +62,9 @@ public class BaseClickhouseData {
}
}
public void BaseVIp() {
void BaseVIp() {
initializeVertexMap(vIpMap);
LOG.info("V_IP resultMap初始化完成");
LOG.info("IP resultMap初始化完成");
String sql = getVIpSql();
long start = System.currentTimeMillis();
try {
@@ -84,9 +89,61 @@ public class BaseClickhouseData {
}
}
public void BaseEFqdnAddressIp() {
void BaseVertexSubscriber(){
initializeVertexMap(vSubscriberMap);
LOG.info("SUBSCRIBER resultMap初始化完成");
String sql = getVertexSubscriberSql();
long start = System.currentTimeMillis();
try {
connection = manger.getConnection();
statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()){
BaseDocument newDoc = getVertexSubscriberDocument(resultSet);
String key = newDoc.getKey();
int i = Math.abs(key.hashCode()) % ApplicationConfig.THREAD_POOL_NUMBER;
HashMap<String, ArrayList<BaseDocument>> documentHashMap = vSubscriberMap.getOrDefault(i, new HashMap<>());
ArrayList<BaseDocument> documentArrayList = documentHashMap.getOrDefault(key, new ArrayList<>());
documentArrayList.add(newDoc);
documentHashMap.put(key,documentArrayList);
}
long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse v_SUBSCRIBER时间：" + (last - start));
}catch (Exception e){
LOG.error(sql + "\n读取clickhouse v_SUBSCRIBER失败");
e.printStackTrace();
}finally {
manger.clear(statement,connection);
}
}
void BaseRelationshipSubscriberLocateIp(){
initializeVertexMap(eSubsciberLocateIpMap);
LOG.info("R_LOCATE_SUBSCRIBER2IP");
String sql = getRelationshipSubsciberLocateIpSql();
long start = System.currentTimeMillis();
try {
connection = manger.getConnection();
statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()){
BaseEdgeDocument newDoc = getRelationshipSubsciberLocateIpDocument(resultSet);
String key = newDoc.getKey();
putMapByHashcode(resultSet, newDoc, eSubsciberLocateIpMap,key);
}
long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse ESubsciberLocateIp时间：" + (last - start));
}catch (Exception e){
LOG.error(sql + "\n读取clickhouse ESubsciberLocateIp失败");
e.printStackTrace();
}finally {
manger.clear(statement,connection);
}
}
void BaseEFqdnAddressIp() {
initializeVertexMap(eFqdnAddressIpMap);
LOG.info("E_ADDRESS_V_FQDN_TO_V_IP resultMap初始化完成");
LOG.info("R_LOCATE_FQDN2IP resultMap初始化完成");
String sql = getEFqdnAddressIpSql();
long start = System.currentTimeMillis();
try {
@@ -96,7 +153,8 @@ public class BaseClickhouseData {
while (resultSet.next()) {
BaseEdgeDocument newDoc = getRelationFqdnAddressIpDocument(resultSet);
putMapByHashcode(resultSet, newDoc, eFqdnAddressIpMap);
String commonSchemaType = resultSet.getString("common_schema_type");
putMapByHashcode(resultSet, newDoc, eFqdnAddressIpMap,commonSchemaType);
}
long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse EFqdnAddressIp时间：" + (last - start));
@@ -107,9 +165,9 @@ public class BaseClickhouseData {
}
}
public void BaseEIpVisitFqdn() {
void BaseEIpVisitFqdn() {
initializeVertexMap(eIpVisitFqdnMap);
LOG.info("E_VISIT_V_IP_TO_V_FQDN resultMap初始化完成");
LOG.info("R_VISIT_IP2FQDN resultMap初始化完成");
String sql = getEIpVisitFqdnSql();
long start = System.currentTimeMillis();
try {
@@ -118,7 +176,8 @@ public class BaseClickhouseData {
ResultSet resultSet = statement.executeQuery(sql);
while (resultSet.next()) {
BaseEdgeDocument newDoc = getRelationIpVisitFqdnDocument(resultSet);
putMapByHashcode(resultSet, newDoc, eIpVisitFqdnMap);
String commonSchemaType = resultSet.getString("common_schema_type");
putMapByHashcode(resultSet, newDoc, eIpVisitFqdnMap,commonSchemaType);
}
long last = System.currentTimeMillis();
LOG.info(sql + "\n读取clickhouse EIpVisitFqdn时间：" + (last - start));

53
IP-learning-graph/src/main/java/cn/ac/iie/dao/UpdateGraphData.java
View File

@@ -1,10 +1,12 @@
package cn.ac.iie.dao;
import cn.ac.iie.config.ApplicationConfig;
import cn.ac.iie.etl.relationship.LocateFqdn2Ip;
import cn.ac.iie.etl.relationship.VisitIp2Fqdn;
import cn.ac.iie.etl.vertex.Fqdn;
import cn.ac.iie.etl.vertex.Ip;
import cn.ac.iie.service.relationship.LocateFqdn2Ip;
import cn.ac.iie.service.relationship.LocateSubscriber2Ip;
import cn.ac.iie.service.relationship.VisitIp2Fqdn;
import cn.ac.iie.service.vertex.Fqdn;
import cn.ac.iie.service.vertex.Ip;
import cn.ac.iie.service.vertex.Subscriber;
import cn.ac.iie.utils.ArangoDBConnect;
import cn.ac.iie.utils.ExecutorThreadPool;
import com.arangodb.entity.BaseDocument;
@@ -16,6 +18,9 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
/**
* 更新图数据库业务类
*/
public class UpdateGraphData {
private static final Logger LOG = LoggerFactory.getLogger(UpdateGraphData.class);
private static ExecutorThreadPool pool = ExecutorThreadPool.getInstance();
@@ -38,10 +43,16 @@ public class UpdateGraphData {
baseClickhouseData.BaseEIpVisitFqdn();
updateRelationIpVisitFqdn();
baseClickhouseData.BaseVertexSubscriber();
updateVertexSubscriber();
baseClickhouseData.BaseRelationshipSubscriberLocateIp();
updateRelationshipSubsciberLocateIp();
}catch (Exception e){
e.printStackTrace();
}finally {
ArangoDBConnect.clean();
arangoManger.clean();
}
long lastC = System.currentTimeMillis();
LOG.info("更新ArangoDb时间："+(lastC - startC));
@@ -63,6 +74,38 @@ public class UpdateGraphData {
}
}
private void updateVertexSubscriber(){
try {
countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
HashMap<String, ArrayList<BaseDocument>> stringArrayListHashMap = BaseClickhouseData.vSubscriberMap.get(i);
LOG.info("vSubscriber baseDocumentHashMap大小：" + stringArrayListHashMap.size());
Subscriber updateVSubscriber = new Subscriber(stringArrayListHashMap, arangoManger, "SUBSCRIBER", BaseArangoData.v_Subscriber_Map,countDownLatch);
updateVSubscriber.run();
}
countDownLatch.await();
LOG.info("---------SUBSCRIBER vertex 更新完毕---------");
}catch (Exception e){
e.printStackTrace();
}
}
private void updateRelationshipSubsciberLocateIp(){
try {
countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
HashMap<String, HashMap<String, BaseEdgeDocument>> baseDocumentHashMap = BaseClickhouseData.eSubsciberLocateIpMap.get(i);
LOG.info("ESubsciberLocateIp baseDocumentHashMap大小：" + baseDocumentHashMap.size());
LocateSubscriber2Ip rLocateSubscriber2IP = new LocateSubscriber2Ip(baseDocumentHashMap, arangoManger, "R_LOCATE_SUBSCRIBER2IP", BaseArangoData.e_Subsciber_Locate_Ip_Map, countDownLatch);
rLocateSubscriber2IP.run();
}
countDownLatch.await();
LOG.info("------------R_LOCATE_SUBSCRIBER2IP relationship 更新完毕----------------");
}catch (Exception e){
e.printStackTrace();
}
}
private void updateVertexIp(){
try {
countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);