21 Commits

Author SHA1 Message Date
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
wanglihui
264afdaa3e 增加radius是否活跃判断条件。 2021-10-25 14:47:22 +08:00
wanglihui
423e9d9b44 统一日志术语,修改表名 2021-09-24 11:06:16 +08:00
wanglihui
d61fbee61a 统一日志术语,修改表名 2021-09-16 11:08:01 +08:00
wanglihui
2f7cceb826 格式化代码 2021-08-02 18:49:43 +08:00
wanglihui
51d2549902 修复class tag无法编译bug。 2021-04-14 14:20:41 +08:00
wanglihui
f0cebd8e1c 格式化代码 2021-03-23 11:26:55 +08:00
wanglihui
b62131dacd 根据自定义arangoRDD重构代码 2020-11-10 16:59:39 +08:00
wanglihui
db5ca9db08 自定义ArangoRDD 2020-10-26 09:54:14 +08:00
wanglihui
c211d99c2e 自定义ArangoRDD 2020-10-23 10:02:28 +08:00
wanglihui
db8e764e00 修改全局变量为本地变量 2020-08-21 18:08:58 +08:00
wanglihui
31e19d7a0f 修改读取arangoDb方式为分页读取。 2020-08-20 09:56:56 +08:00
wanglihui
5be662f898 修改读取arangoDb方式为分页读取。 2020-08-20 09:56:39 +08:00
38 changed files with 2054 additions and 735 deletions

View File

@@ -39,6 +39,12 @@
<version>1.2.1</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>com.arangodb</groupId>
<artifactId>arangodb-java-driver</artifactId>

View File

@@ -14,7 +14,7 @@ public class ApplicationConfig {
public static final Integer ARANGODB_BATCH = ConfigUtils.getIntProperty( "arangoDB.batch");
public static final Integer UPDATE_ARANGO_BATCH = ConfigUtils.getIntProperty("update.arango.batch");
public static final String ARANGODB_READ_LIMIT = ConfigUtils.getStringProperty("arangoDB.read.limit");
public static final Long ARANGODB_READ_LIMIT = ConfigUtils.getLongProperty("arangoDB.read.limit");
public static final Integer THREAD_POOL_NUMBER = ConfigUtils.getIntProperty( "thread.pool.number");
public static final Integer THREAD_AWAIT_TERMINATION_TIME = ConfigUtils.getIntProperty( "thread.await.termination.time");

View File

@@ -41,9 +41,11 @@ public class BaseArangoData {
map.put(i,new ConcurrentHashMap<>());
}
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
long[] timeRange = getTimeRange(table);
// long[] timeRange = getTimeRange(table);
Long countTotal = getCountTotal(table);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
String sql = getQuerySql(timeRange, i, table);
// String sql = getQuerySql(timeRange, i, table);
String sql = getQuerySql(countTotal, i, table);
ReadHistoryArangoData<T> readHistoryArangoData =
new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
threadPool.executor(readHistoryArangoData);
@@ -56,6 +58,32 @@ public class BaseArangoData {
}
}
public <T extends BaseDocument> ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> readHistoryData(String table, Class<T> type){
ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map = new ConcurrentHashMap<>();
try {
LOG.info("开始更新"+table);
long start = System.currentTimeMillis();
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++){
map.put(i,new ConcurrentHashMap<>());
}
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
Long countTotal = getCountTotal(table);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
String sql = getQuerySql(countTotal, i, table);
ReadHistoryArangoData<T> readHistoryArangoData =
new ReadHistoryArangoData<>(arangoDBConnect, sql, map,type,table,countDownLatch);
threadPool.executor(readHistoryArangoData);
}
countDownLatch.await();
long last = System.currentTimeMillis();
LOG.info("读取"+table+" arangoDB 共耗时:"+(last-start));
}catch (Exception e){
e.printStackTrace();
LOG.error("读取历史数据失败 "+e.toString());
}
return map;
}
private long[] getTimeRange(String table){
long minTime = 0L;
long maxTime = 0L;
@@ -81,6 +109,33 @@ public class BaseArangoData {
}
private Long getCountTotal(String table){
long start = System.currentTimeMillis();
Long cnt = 0L;
String sql = "RETURN LENGTH("+table+")";
try {
ArangoCursor<Long> longs = arangoDBConnect.executorQuery(sql, Long.class);
while (longs.hasNext()){
cnt = longs.next();
}
}catch (Exception e){
LOG.error(sql +"执行异常");
}
long last = System.currentTimeMillis();
LOG.info(sql+" 结果:"+cnt+" 执行时间:"+(last-start));
return cnt;
}
private String getQuerySql(Long cnt,int threadNumber, String table){
long sepNum = cnt / ApplicationConfig.THREAD_POOL_NUMBER + 1;
long offsetNum = threadNumber * sepNum;
if (sepNum > ApplicationConfig.ARANGODB_READ_LIMIT){
sepNum = ApplicationConfig.ARANGODB_READ_LIMIT;
}
return "FOR doc IN " + table + " limit "+offsetNum+","+sepNum+" RETURN doc";
}
private String getQuerySql(long[] timeRange,int threadNumber,String table){
long minTime = timeRange[0];
long maxTime = timeRange[1];

View File

@@ -73,5 +73,51 @@ public class BaseClickhouseData {
}
}
public <T extends BaseDocument> HashMap<Integer, HashMap<String,ArrayList<T>>> baseDocFromCk(Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc){
long start = System.currentTimeMillis();
HashMap<Integer, HashMap<String, ArrayList<T>>> newDataMap = initializeMap();
if (newDataMap == null){
return null;
}
String sql = getSqlSupplier.get();
try {
connection = manger.getConnection();
statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
int i = 0;
while (resultSet.next()) {
T newDoc = formatResultFunc.apply(resultSet);
if (newDoc != null) {
i+=1;
putMapByHashcode(newDoc, newDataMap);
}
}
long last = System.currentTimeMillis();
LOG.info(sql + "\n读取"+i+"条数据,运行时间:" + (last - start));
}catch (Exception e){
e.printStackTrace();
LOG.error("获取原始数据失败 "+e.toString());
}finally {
manger.clear(statement,connection);
}
return newDataMap;
}
private <T extends BaseDocument> HashMap<Integer, HashMap<String,ArrayList<T>>> initializeMap(){
try {
HashMap<Integer, HashMap<String, ArrayList<T>>> newDataMap = new HashMap<>();
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++) {
newDataMap.put(i, new HashMap<>());
}
return newDataMap;
}catch (Exception e){
e.printStackTrace();
LOG.error("数据初始化失败 "+e.toString());
return null;
}
}
}

View File

@@ -44,8 +44,8 @@ public class UpdateGraphData {
long start = System.currentTimeMillis();
try {
updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN", Fqdn.class,BaseDocument.class,
ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
// updateDocument(newVertexFqdnMap, historyVertexFqdnMap, "FQDN", Fqdn.class,BaseDocument.class,
// ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
updateDocument(newVertexIpMap,historyVertexIpMap,"IP", Ip.class,BaseDocument.class,
ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);
@@ -75,6 +75,41 @@ public class UpdateGraphData {
}
}
public void updateArango2(){
long start = System.currentTimeMillis();
try {
// updateDocument("FQDN", Fqdn.class,BaseDocument.class,
// ReadClickhouseData::getVertexFqdnSql,ReadClickhouseData::getVertexFqdnDocument);
updateDocument("IP", Ip.class,BaseDocument.class,
ReadClickhouseData::getVertexIpSql,ReadClickhouseData::getVertexIpDocument);
updateDocument("SUBSCRIBER", Subscriber.class,BaseDocument.class,
ReadClickhouseData::getVertexSubscriberSql,ReadClickhouseData::getVertexSubscriberDocument);
updateDocument("R_LOCATE_FQDN2IP", LocateFqdn2Ip.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipFqdnAddressIpSql,ReadClickhouseData::getRelationFqdnAddressIpDocument);
// updateDocument("R_VISIT_IP2FQDN",
// VisitIp2Fqdn.class,BaseEdgeDocument.class,
// ReadClickhouseData::getRelationshipIpVisitFqdnSql,ReadClickhouseData::getRelationIpVisitFqdnDocument);
updateDocument("R_LOCATE_SUBSCRIBER2IP",
LocateSubscriber2Ip.class,BaseEdgeDocument.class,
ReadClickhouseData::getRelationshipSubsciberLocateIpSql,ReadClickhouseData::getRelationshipSubsciberLocateIpDocument);
long last = System.currentTimeMillis();
LOG.info("iplearning application运行完毕用时"+(last - start));
}catch (Exception e){
e.printStackTrace();
}finally {
arangoManger.clean();
pool.shutdown();
}
}
private <T extends BaseDocument> void updateDocument(HashMap<Integer, HashMap<String, ArrayList<T>>> newMap,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
@@ -116,5 +151,45 @@ public class UpdateGraphData {
}
}
private <T extends BaseDocument> void updateDocument(String collection,
Class<? extends Document<T>> taskType,
Class<T> docmentType,
Supplier<String> getSqlSupplier,
Function<ResultSet,T> formatResultFunc){
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyData = baseArangoData.readHistoryData(collection, docmentType);
LOG.info(collection+" 读取clickhouse,封装结果集");
HashMap<Integer, HashMap<String, ArrayList<T>>> newData = baseClickhouseData.baseDocFromCk(getSqlSupplier, formatResultFunc);
try {
LOG.info(collection+" 开始更新");
long start = System.currentTimeMillis();
CountDownLatch countDownLatch = new CountDownLatch(ApplicationConfig.THREAD_POOL_NUMBER);
for (int i = 0; i < ApplicationConfig.THREAD_POOL_NUMBER; i++){
HashMap<String, ArrayList<T>> tmpNewMap = newData.get(i);
ConcurrentHashMap<String, T> tmpHisMap = historyData.get(i);
Constructor constructor = taskType.getConstructor(
HashMap.class,
ArangoDBConnect.class,
String.class,
ConcurrentHashMap.class,
CountDownLatch.class);
Document docTask = (Document)constructor.newInstance(tmpNewMap, arangoManger, collection, tmpHisMap, countDownLatch);
pool.executor(docTask);
}
countDownLatch.await();
long last = System.currentTimeMillis();
LOG.info(collection+" 更新完毕,共耗时:"+(last-start));
}catch (Exception e){
e.printStackTrace();
LOG.error("更新"+collection+"失败!!"+e.toString());
}finally {
newData.clear();
historyData.clear();
}
}
}

View File

@@ -257,40 +257,40 @@ 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.connection_record_log 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.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host";
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";
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) as common_link_info,'client' as ip_type FROM tsg_galaxy_v3.connection_record_log 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) as common_link_info,'server' as ip_type FROM tsg_galaxy_v3.connection_record_log where " + where + " group by IP";
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";
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.connection_record_log 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.connection_record_log WHERE " + where + " and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip";
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";
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.connection_record_log 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.connection_record_log WHERE common_schema_type = 'SSL' GROUP BY ssl_sni,common_client_ip";
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";
return "SELECT * FROM ((" + sslSql + ") UNION ALL (" + httpSql + "))WHERE FQDN != ''";
}
public static String getVertexSubscriberSql() {
String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
return "SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id";
return "SELECT common_subscriber_id,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME FROM radius_record WHERE" + where + " GROUP BY common_subscriber_id";
}
public static String getRelationshipSubsciberLocateIpSql() {
String where = " common_recv_time >= " + minTime + " AND common_recv_time < " + maxTime + " AND common_subscriber_id != '' AND radius_framed_ip != '' AND radius_packet_type = 4 AND radius_acct_status_type = 1";
return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record_log WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip";
return "SELECT common_subscriber_id,radius_framed_ip,MAX(common_recv_time) as LAST_FOUND_TIME,MIN(common_recv_time) as FIRST_FOUND_TIME,COUNT(*) as COUNT_TOTAL FROM radius_record WHERE" + where + " GROUP BY common_subscriber_id,radius_framed_ip";
}
private static long[] getTimeLimit() {

View File

@@ -33,10 +33,10 @@ public class Ip extends Vertex {
}
private void updateIpByType(BaseDocument newDocument, BaseDocument historyDocument) {
putSumAttribute(newDocument, historyDocument, "CLIENT_SESSION_COUNT");
putSumAttribute(newDocument, historyDocument, "CLIENT_BYTES_SUM");
putSumAttribute(newDocument, historyDocument, "SERVER_SESSION_COUNT");
putSumAttribute(newDocument, historyDocument, "SERVER_BYTES_SUM");
// putSumAttribute(newDocument, historyDocument, "CLIENT_SESSION_COUNT");
// putSumAttribute(newDocument, historyDocument, "CLIENT_BYTES_SUM");
// putSumAttribute(newDocument, historyDocument, "SERVER_SESSION_COUNT");
// putSumAttribute(newDocument, historyDocument, "SERVER_BYTES_SUM");
}
}

View File

@@ -6,7 +6,7 @@ public class IpLearningApplicationTest {
public static void main(String[] args) {
UpdateGraphData updateGraphData = new UpdateGraphData();
updateGraphData.updateArango();
updateGraphData.updateArango2();
}
}

View File

@@ -1,15 +1,13 @@
#arangoDB参数配置
arangoDB.host=192.168.40.182
#arangoDB.host=192.168.40.224
arangoDB.host=192.168.44.12
arangoDB.port=8529
arangoDB.user=upsert
arangoDB.password=ceiec2018
arangoDB.DB.name=ip-learning-test
#arangoDB.DB.name=tsg_galaxy_v3
arangoDB.user=root
arangoDB.password=ceiec2019
arangoDB.DB.name=tsg_galaxy_v3
arangoDB.batch=100000
arangoDB.ttl=3600
arangoDB.read.limit=
arangoDB.read.limit=10000000
update.arango.batch=10000
thread.pool.number=10

View File

@@ -1,9 +1,7 @@
drivers=ru.yandex.clickhouse.ClickHouseDriver
mdb.user=default
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
db.id=192.168.44.67:8123/tsg_galaxy_v3?socket_timeout=3600000
mdb.password=ceiec2019
initialsize=1
minidle=1
maxactive=50

View File

@@ -0,0 +1,46 @@
package cn.ac.iie;
import cn.ac.iie.dao.BaseClickhouseData;
import cn.ac.iie.service.ingestion.ReadClickhouseData;
import com.arangodb.entity.BaseDocument;
import com.arangodb.entity.BaseEdgeDocument;
import org.junit.Test;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Set;
public class TestReadClickhouse {
private static BaseClickhouseData baseClickhouseData = new BaseClickhouseData();
@Test
public void testReadFqdnFromCk(){
HashMap<Integer, HashMap<String, ArrayList<BaseDocument>>> newData =
baseClickhouseData.baseDocFromCk(ReadClickhouseData::getVertexFqdnSql,
ReadClickhouseData::getVertexFqdnDocument);
printMap(newData);
}
@Test
public void testReadFqdnLocIpFromCk(){
HashMap<Integer, HashMap<String, ArrayList<BaseEdgeDocument>>> map =
baseClickhouseData.baseDocFromCk(ReadClickhouseData::getRelationshipFqdnAddressIpSql,
ReadClickhouseData::getRelationFqdnAddressIpDocument);
printMap(map);
}
private<T extends BaseDocument> void printMap(HashMap<Integer, HashMap<String, ArrayList<T>>> newData){
HashMap<String, ArrayList<T>> map = newData.get(1);
Set<String> strings = map.keySet();
for (String key:strings){
ArrayList<T> baseDocuments = map.get(key);
System.out.println(baseDocuments.get(0));
}
}
}

View File

@@ -7,82 +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>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.8</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>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.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>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>19.0</version>
</dependency>
<dependency>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<version>2.15.2</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>
</dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.2.3</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>
<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>

View File

@@ -1,103 +0,0 @@
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<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexFqdnMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexIpMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseDocument>> historyVertexSubscriberMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnAddressIpMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationIpVisitFqdnMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationFqdnSameFqdnMap = new ConcurrentHashMap<>();
public static ConcurrentHashMap<Integer, ConcurrentHashMap<String, BaseEdgeDocument>> historyRelationSubsciberLocateIpMap = new ConcurrentHashMap<>();
private static ArangoDBConnect arangoDBConnect = ArangoDBConnect.getInstance();
private ExecutorThreadPool threadPool = ExecutorThreadPool.getInstance();
public <T extends BaseDocument> void readHistoryData(String table,
ConcurrentHashMap<Integer, ConcurrentHashMap<String, T>> historyMap,
Class<T> 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<T> 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<BaseDocument> 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";
}
}

View File

@@ -1,125 +0,0 @@
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<T extends BaseDocument> 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<String> 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<Integer,ConcurrentHashMap<String, T>> map;
private Class<T> type;
private String table;
private CountDownLatch countDownLatch;
public ReadHistoryArangoData(ArangoDBConnect arangoConnect,
String query,
ConcurrentHashMap<Integer,ConcurrentHashMap<String, T>> map,
Class<T> 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<T> docs = arangoConnect.executorQuery(query, type);
if (docs != null) {
List<T> 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<String, T> 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<Long> cntRecent = (ArrayList<Long>) 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<String> distCip = (ArrayList<String>) doc.getAttribute("DIST_CIP");
ArrayList<Long> distCipTs = (ArrayList<Long>) 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);
}
}

View File

@@ -22,11 +22,12 @@ 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(){
private static void getArangoDatabase() {
arangoDB = new ArangoDB.Builder()
.maxConnections(ApplicationConfig.THREAD_POOL_NUMBER())
.host(ApplicationConfig.ARANGODB_HOST(), ApplicationConfig.ARANGODB_PORT())
@@ -35,82 +36,81 @@ public class ArangoDBConnect {
.build();
}
public static synchronized ArangoDBConnect getInstance(){
if (null == conn){
public static synchronized ArangoDBConnect getInstance() {
if (null == conn) {
conn = new ArangoDBConnect();
}
return conn;
}
private ArangoDatabase getDatabase(){
private ArangoDatabase getDatabase() {
return arangoDB.db(ApplicationConfig.ARANGODB_DB_NAME());
}
public void clean(){
public void clean() {
try {
if (arangoDB != null){
if (arangoDB != null) {
arangoDB.shutdown();
}
}catch (Exception e){
} catch (Exception e) {
e.printStackTrace();
}
}
public <T> ArangoCursor<T> executorQuery(String query,Class<T> type){
public <T> ArangoCursor<T> executorQuery(String query, Class<T> type) {
ArangoDatabase database = getDatabase();
Map<String, Object> bindVars = new MapBuilder().get();
AqlQueryOptions options = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL());
try {
return database.query(query, bindVars, options, type);
}catch (Exception e){
} catch (Exception e) {
e.printStackTrace();
return null;
}finally {
} finally {
bindVars.clear();
}
}
@Deprecated
public <T> void insertAndUpdate(ArrayList<T> docInsert,ArrayList<T> docUpdate,String collectionName){
public <T> void insertAndUpdate(ArrayList<T> docInsert, ArrayList<T> docUpdate, String collectionName) {
ArangoDatabase database = getDatabase();
try {
ArangoCollection collection = database.collection(collectionName);
if (!docInsert.isEmpty()){
if (!docInsert.isEmpty()) {
collection.importDocuments(docInsert);
}
if (!docUpdate.isEmpty()){
if (!docUpdate.isEmpty()) {
collection.replaceDocuments(docUpdate);
}
}catch (Exception e){
} catch (Exception e) {
System.out.println("更新失败");
e.printStackTrace();
}finally {
} finally {
docInsert.clear();
docInsert.clear();
}
}
public <T> void overwrite(ArrayList<T> docOverwrite,String collectionName){
public <T> void overwrite(ArrayList<T> docOverwrite, String collectionName) {
ArangoDatabase database = getDatabase();
try {
ArangoCollection collection = database.collection(collectionName);
if (!docOverwrite.isEmpty()){
if (!docOverwrite.isEmpty()) {
DocumentCreateOptions documentCreateOptions = new DocumentCreateOptions();
documentCreateOptions.overwrite(true);
documentCreateOptions.silent(true);
MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
for (ErrorEntity errorEntity:errors){
LOG.warn("写入arangoDB异常"+errorEntity.getErrorMessage());
for (ErrorEntity errorEntity : errors) {
LOG.warn("写入arangoDB异常" + errorEntity.getErrorMessage());
}
}
}catch (Exception e){
System.out.println("更新失败:"+e.toString());
}finally {
} catch (Exception e) {
LOG.error("更新失败:" + e.toString());
} finally {
docOverwrite.clear();
}
}
}

View File

@@ -1,45 +1,45 @@
#spark任务配置
spark.sql.shuffle.partitions=5
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
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.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=111111
spark.read.clickhouse.numPartitions=144
spark.read.clickhouse.fetchsize=10000
spark.read.clickhouse.partitionColumn=common_recv_time
spark.read.clickhouse.password=galaxy2019
spark.read.clickhouse.numPartitions=5
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.182
arangoDB.host=192.168.44.83
arangoDB.port=8529
arangoDB.user=upsert
arangoDB.password=ceiec2018
#arangoDB.DB.name=insert_iplearn_index
arangoDB.DB.name=ip-learning-test-0
arangoDB.user=root
arangoDB.password=galaxy_2019
arangoDB.DB.name=tsg_galaxy_v3
arangoDB.ttl=3600
thread.pool.number=5
thread.pool.number=10
#读取clickhouse时间范围方式0读取过去一小时1指定时间范围
clickhouse.time.limit.type=0
read.clickhouse.max.time=1571245220
read.clickhouse.min.time=1571245210
clickhouse.time.limit.type=1
read.clickhouse.max.time=1634902508
read.clickhouse.min.time=1631759985
#读取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=10800
update.interval=3600
arangodb.total.num=20000000
#读取radius时间范围,与radius任务执行周期一致,单位:分钟
read.radius.granularity=-30
vsys.id=1

View File

@@ -4,20 +4,15 @@ log4j.logger.org.apache.http.wire=OFF
#Log4j
log4j.rootLogger=info,console,file
# <20><><EFBFBD><EFBFBD>̨<EFBFBD><CCA8>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
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
# <20>ļ<EFBFBD><C4BC><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
log4j.appender.file.Threshold=warn
log4j.appender.file.encoding=UTF-8
log4j.appender.file.Append=true
<><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ز<EFBFBD><D8B2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6>Ŀ<EFBFBD><C4BF>
#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

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")
@@ -20,6 +22,8 @@ object ApplicationConfig {
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 SPARK_READ_CLICKHOUSE_SESSION_TABLE: String = config.getString("spark.read.clickhouse.session.table")
val SPARK_READ_CLICKHOUSE_RADIUS_TABLE: String = config.getString("spark.read.clickhouse.radius.table")
val ARANGODB_HOST: String= config.getString("arangoDB.host")
val ARANGODB_PORT: Int = config.getInt("arangoDB.port")
@@ -36,16 +40,14 @@ object ApplicationConfig {
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")
val ARANGODB_TOTAL_NUM: Long = config.getLong("arangodb.total.num")
val READ_RADIUS_GRANULARITY: Int = config.getInt("read.radius.granularity")
}

View File

@@ -0,0 +1,22 @@
package cn.ac.iie.dao
import cn.ac.iie.config.ApplicationConfig
import cn.ac.iie.spark.ArangoSpark
import cn.ac.iie.spark.rdd.{ArangoRdd, ReadOptions}
import cn.ac.iie.utils.SparkSessionUtil.sparkContext
import org.slf4j.LoggerFactory
import scala.reflect.ClassTag
object BaseArangoData {
private val LOG = LoggerFactory.getLogger(BaseArangoData.getClass)
private val options = ReadOptions(ApplicationConfig.ARANGODB_DB_NAME)
def loadArangoRdd[T: ClassTag](name:String): ArangoRdd[T] ={
val value = ArangoSpark.load[T](sparkContext, name, options)
LOG.warn(s"读取$name arangoDb:${value.count()}")
value
}
}

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
@@ -11,7 +14,7 @@ object BaseClickhouseData {
val currentHour: Long = System.currentTimeMillis / (60 * 60 * 1000) * 60 * 60
private val timeLimit: (Long, Long) = getTimeLimit
private def initClickhouseData(sql:String): Unit ={
private def initClickhouseData(sql: String): DataFrame = {
val dataFrame: DataFrame = spark.read.format("jdbc")
.option("url", ApplicationConfig.SPARK_READ_CLICKHOUSE_URL)
@@ -24,174 +27,179 @@ object BaseClickhouseData {
.option("lowerBound", timeLimit._2)
.option("upperBound", timeLimit._1)
.option("fetchsize", ApplicationConfig.SPARK_READ_CLICKHOUSE_FETCHSIZE)
.option("socket_timeout",ApplicationConfig.CLICKHOUSE_SOCKET_TIMEOUT)
.option("socket_timeout", ApplicationConfig.CLICKHOUSE_SOCKET_TIMEOUT)
.load()
dataFrame.printSchema()
dataFrame.createOrReplaceGlobalTempView("dbtable")
dataFrame
}
def loadConnectionDataFromCk(): Unit ={
def getVertexFqdnDf: DataFrame = {
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
|(SELECT
| FQDN,MAX( LAST_FOUND_TIME ) AS LAST_FOUND_TIME,MIN( FIRST_FOUND_TIME ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
|FROM
| ((SELECT
| ssl_sni AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
| WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_vsys_id
| )UNION ALL
| (SELECT
| http_host AS FQDN,MAX( common_recv_time ) AS LAST_FOUND_TIME,MIN( common_recv_time ) AS FIRST_FOUND_TIME,common_vsys_id AS VSYS_ID
| FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
| WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_vsys_id))
|GROUP BY FQDN,VSYS_ID HAVING FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)
initClickhouseData(sql)
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
private def loadRadiusDataFromCk(): Unit ={
def getVertexIpDf: DataFrame = {
val where = "common_recv_time >= " + timeLimit._2 + " AND common_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,
|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,
|'client' as ip_type
|,common_vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_client_ip,common_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,
|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,
|'server' as ip_type
|,common_vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|where $where
|group by common_server_ip,common_vsys_id))) as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
def getRelationFqdnLocateIpDf: DataFrame = {
val where = "common_recv_time >= " + timeLimit._2 + " AND common_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,common_vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'SSL' GROUP BY ssl_sni,common_server_ip,common_vsys_id)
|UNION ALL
|(SELECT http_host AS FQDN,common_server_ip,MAX(common_recv_time) AS LAST_FOUND_TIME,MIN(common_recv_time) AS FIRST_FOUND_TIME,COUNT(*) AS COUNT_TOTAL,
|toString(groupUniqArray(${ApplicationConfig.DISTINCT_CLIENT_IP_NUM})(common_client_ip)) AS DIST_CIP_RECENT,'HTTP' AS schema_type,common_vsys_id AS VSYS_ID
|FROM ${ApplicationConfig.SPARK_READ_CLICKHOUSE_SESSION_TABLE}
|WHERE $where and common_schema_type = 'HTTP' GROUP BY http_host,common_server_ip,common_vsys_id))
|WHERE FQDN != '') as dbtable
""".stripMargin
LOG.warn(sql)
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
def getRelationSubidLocateIpDf: DataFrame = {
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
| 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,common_recv_time
|FROM
| tsg_galaxy_v3.radius_record_log
|WHERE
| $where) as dbtable
|(
|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,common_vsys_id
|) as dbtable
""".stripMargin
LOG.warn(sql)
initClickhouseData(sql)
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
def getVertexFqdnDf: DataFrame ={
loadConnectionDataFromCk()
def getVertexSubidDf: DataFrame = {
val where =
s"""
| 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 =
"""
|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 != ''
s"""
|(
|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)
val vertexFqdnDf = spark.sql(sql)
vertexFqdnDf.printSchema()
vertexFqdnDf
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
def getVertexIpDf: DataFrame ={
loadConnectionDataFromCk()
def getVertexFramedIpDf: DataFrame = {
val where =
s"""
| 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 =
"""
|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
| )
| )
s"""
|(
|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)
val vertexIpDf = spark.sql(sql)
vertexIpDf.printSchema()
vertexIpDf
val frame = initClickhouseData(sql)
frame.printSchema()
frame
}
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 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)
}
private def getTimeLimit: (Long,Long) ={
def main(args: Array[String]): Unit = {
println(getRadiusTimeRange)
println(getRadiusTimeRange._2 - getRadiusTimeRange._1)
}
private def getTimeLimit: (Long, Long) = {
var maxTime = 0L
var minTime = 0L
ApplicationConfig.CLICKHOUSE_TIME_LIMIT_TYPE match {

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

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

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

View File

@@ -1,42 +1,57 @@
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.dao.{BaseArangoData, BaseClickhouseData}
import cn.ac.iie.spark.partition.CustomPartitioner
import cn.ac.iie.spark.rdd.ArangoRdd
import com.arangodb.entity.{BaseDocument, BaseEdgeDocument}
import org.apache.spark.rdd.RDD
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[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 mergeVertexFqdn(): RDD[(String, (Option[BaseDocument], Row))] = {
val fqdnRddRow: RDD[(String, Row)] = BaseClickhouseData.getVertexFqdnDf
.repartition().rdd.filter(row => isDomain(row.getAs[String](0))).map(row => {
(row.getAs[String]("FQDN"), row)
}) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
val fqdnRddDoc: ArangoRdd[BaseDocument] = BaseArangoData.loadArangoRdd[BaseDocument]("FQDN")
fqdnRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnRddRow)
}
def mergeVertexIp(): RDD[Row]={
def mergeVertexIp(): RDD[(String, (Option[BaseDocument], Row))] = {
val vertexIpDf = BaseClickhouseData.getVertexIpDf
val frame = vertexIpDf.groupBy("IP").agg(
val frame = vertexIpDf.repartition().groupBy("IP", "VSYS_ID").agg(
min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
max("LAST_FOUND_TIME").alias("LAST_FOUND_TIME"),
collect_list("SESSION_COUNT").alias("SESSION_COUNT_LIST"),
collect_list("BYTES_SUM").alias("BYTES_SUM_LIST"),
collect_list("ip_type").alias("ip_type_list")
collect_list("ip_type").alias("ip_type_list"),
last("common_link_info").alias("common_link_info")
)
val values = frame.rdd.map(row => (row.get(0), row))
.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
values
val ipRddRow = frame.rdd.map(row => {
val vsysId = row.getAs[Long]("VSYS_ID")
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)
}
def mergeRelationFqdnLocateIp(): RDD[Row] ={
val frame = BaseClickhouseData.getRelationFqdnLocateIpDf.filter(row => isDomain(row.getAs[String]("FQDN")))
.groupBy("FQDN", "common_server_ip")
def mergeRelationFqdnLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
val frame = BaseClickhouseData.getRelationFqdnLocateIpDf
.repartition().filter(row => isDomain(row.getAs[String]("FQDN")))
.groupBy("FQDN", "common_server_ip", "VSYS_ID")
.agg(
min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
max("LAST_FOUND_TIME").alias("LAST_FOUND_TIME"),
@@ -44,28 +59,66 @@ object MergeDataFrame {
collect_list("schema_type").alias("schema_type_list"),
collect_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT")
)
frame.rdd.map(row => {
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)
(key,row)
}).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
val vsysId = row.getAs[Long]("VSYS_ID")
val key = fqdn.concat("-" + serverIp + "-" + vsysId)
(key, row)
}) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
val fqdnLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_FQDN2IP")
fqdnLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnLocIpRddRow)
}
def mergeRelationSubidLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
val subidLocIpRddRow = BaseClickhouseData.getRelationSubidLocateIpDf
.repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
.rdd.map(row => {
val commonSubscriberId = row.getAs[String]("common_subscriber_id")
val ip = row.getAs[String]("radius_framed_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")
subidLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(subidLocIpRddRow)
}
def mergeVertexSubid(): RDD[(String, (Option[BaseDocument], Row))] = {
val subidRddRow = BaseClickhouseData.getVertexSubidDf
.repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
.rdd.map(row => {
val commonSubscriberId = row.getAs[String]("common_subscriber_id")
(commonSubscriberId, row)
}).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
val subidRddDoc = BaseArangoData.loadArangoRdd[BaseDocument]("SUBSCRIBER")
subidRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(subidRddRow)
}
def mergeVertexFrameIp: RDD[Row] = {
val values = BaseClickhouseData.getVertexFramedIpDf
.repartition(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)
.rdd.map(row => {
val ip = row.getAs[String]("radius_framed_ip")
(ip, row)
}).partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS)).values
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){
val fqdnArr = fqdn.split(":")(0).split("\\.")
if (fqdnArr.length != 4) {
return true
}
for (f <- fqdnArr) {
@@ -83,6 +136,7 @@ object MergeDataFrame {
LOG.error("解析域名 " + fqdn + " 失败:\n" + e.toString)
}
false
}
}

View File

@@ -1,39 +1,50 @@
package cn.ac.iie.service.update
import java.lang
import java.util
import scala.collection.JavaConversions._
import cn.ac.iie.config.ApplicationConfig
import cn.ac.iie.service.read.ReadHistoryArangoData
import cn.ac.iie.dao.BaseClickhouseData
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")
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
def updateMaxAttribute(hisDoc: BaseDocument, newAttribute: Long, attributeName: String): Unit = {
if (hisDoc.getProperties.containsKey(attributeName)) {
var hisAttritube = hisDoc.getAttribute(attributeName).toString.toLong
if (newAttribute > hisAttritube) {
hisAttritube = newAttribute
}
hisDoc.addAttribute(attributeName, hisAttritube)
}
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 updateSumAttribute(hisDoc: BaseDocument, newAttribute: Long, attributeName: String): Unit = {
if (hisDoc.getProperties.containsKey(attributeName)) {
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) ={
def replaceAttribute(hisDoc: BaseDocument, newAttribute: String, attributeName: String): Unit = {
// hisDoc.addAttribute(attributeName,newAttribute)
hisDoc.updateAttribute(attributeName, newAttribute)
}
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){
if (ipTypeList.length == sessionCountList.length && ipTypeList.length == bytesSumList.length) {
sessionCountList.zip(bytesSumList).zip(ipTypeList).foreach(t => {
t._2 match {
case "server" =>
@@ -48,75 +59,99 @@ object UpdateDocHandler {
(serverSessionCount, serverBytesSum, clientSessionCount, clientBytesSum)
}
def separateAttributeByProtocol(schemaTypeList:ofRef[AnyRef],countTotalList:ofRef[AnyRef]): Map[String, Long] ={
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
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)){
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 updateProtocolAttritube(hisDoc: BaseEdgeDocument, protocolMap: Map[String, Long]): Unit = {
if (hisDoc.getProperties.containsKey("PROTOCOL_TYPE")) {
var protocolType = hisDoc.getAttribute("PROTOCOL_TYPE").toString
protocolMap.foreach((t: (String, Long)) => {
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 = hisDoc.getAttribute(cntRecentName).asInstanceOf[Array[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 ={
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)
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)
cntRecent.update(0, t._2)
doc.addAttribute(cntTotalName, t._2)
doc.addAttribute(cntRecentName, cntRecent)
})
doc.addAttribute("PROTOCOL_TYPE",protocolTypeBuilder.toString().replaceFirst(",",""))
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 updateProtocolDocument(doc: BaseEdgeDocument): Unit = {
if (doc.getProperties.containsKey("PROTOCOL_TYPE")) {
for (protocol <- PROTOCOL_SET) {
val protocolRecent = protocol + "_CNT_RECENT"
val cntRecent: util.ArrayList[Long] = doc.getAttribute(protocolRecent).asInstanceOf[util.ArrayList[Long]]
val cntRecentsSrc = cntRecent.toArray().map(_.toString.toLong)
val cntRecentsDst = new Array[Long](24)
System.arraycopy(cntRecentsSrc, 0, cntRecentsDst, 1, cntRecentsSrc.length - 1)
cntRecentsDst(0) = 0L
doc.addAttribute(protocolRecent, cntRecentsDst)
}
}
}
def putDistinctIp(doc:BaseEdgeDocument,newDistinctIp:Array[String]): Unit ={
def mergeDistinctIp(distCipRecent: ofRef[String]): Array[String] = {
distCipRecent.flatMap(str => {
str.replaceAll("\\[", "")
.replaceAll("\\]", "")
.replaceAll("\\'", "")
.split(",")
}).distinct.take(ApplicationConfig.DISTINCT_CLIENT_IP_NUM).toArray
}
def putDistinctIp(doc: BaseEdgeDocument, newDistinctIp: Array[String]): Unit = {
val map = newDistinctIp.map(ip => {
(ip, ReadHistoryArangoData.currentHour)
(ip, BaseClickhouseData.currentHour)
}).toMap
doc.addAttribute("DIST_CIP",map.keys.toArray)
doc.addAttribute("DIST_CIP_TS",map.values.toArray)
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)
def updateDistinctIp(hisDoc: BaseEdgeDocument, newDistinctIp: Array[String]): Unit = {
if (hisDoc.getProperties.containsKey("DIST_CIP") && hisDoc.getProperties.containsKey("DIST_CIP_TS")) {
val hisDistCip = hisDoc.getAttribute("DIST_CIP").asInstanceOf[util.ArrayList[String]]
val hisDistCipTs = hisDoc.getAttribute("DIST_CIP_TS").asInstanceOf[util.ArrayList[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, BaseClickhouseData.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)
}
}
}

View File

@@ -1,17 +1,12 @@
package cn.ac.iie.service.update
import java.util
import java.util.concurrent.ConcurrentHashMap
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, ExecutorThreadPool, SparkSessionUtil}
import cn.ac.iie.utils.SparkSessionUtil.spark
import cn.ac.iie.utils.{ArangoDBConnect, SparkSessionUtil}
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
@@ -19,44 +14,67 @@ import org.slf4j.LoggerFactory
import scala.collection.mutable.WrappedArray.ofRef
object UpdateDocument {
private val pool = ExecutorThreadPool.getInstance
private val arangoManger: ArangoDBConnect = ArangoDBConnect.getInstance()
private val LOG = LoggerFactory.getLogger(UpdateDocument.getClass)
private val baseArangoData = new BaseArangoData()
def update(): Unit = {
def ipLearning(): Unit = {
try {
updateDocument("FQDN", historyVertexFqdnMap, getVertexFqdnRow, classOf[BaseDocument], mergeVertexFqdn)
updateDocument("IP", historyVertexIpMap, getVertexIpRow, classOf[BaseDocument], mergeVertexIp)
updateDocument("R_LOCATE_FQDN2IP", historyRelationFqdnAddressIpMap, getRelationFqdnLocateIpRow, classOf[BaseEdgeDocument], mergeRelationFqdnLocateIp)
updateDocument("R_LOCATE_FQDN2IP", getRelationFqdnLocateIpRow, mergeRelationFqdnLocateIp)
} catch {
case e: Exception => e.printStackTrace()
} finally {
pool.shutdown()
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 {
arangoManger.clean()
SparkSessionUtil.closeSpark()
System.exit(0)
}
}
private def updateDocument[T <: BaseDocument](collName: String,
historyMap: ConcurrentHashMap[Integer, ConcurrentHashMap[String, T]],
getDocumentRow: (Row, ConcurrentHashMap[String, T]) => T,
clazz: Class[T],
getNewDataRdd: () => RDD[Row]
getDocumentRow: ((String, (Option[T], Row))) => T,
getJoinRdd: () => RDD[(String, (Option[T], Row))]
): Unit = {
baseArangoData.readHistoryData(collName, historyMap, clazz)
val hisBc = spark.sparkContext.broadcast(historyMap)
try {
val start = System.currentTimeMillis()
val newDataRdd = getNewDataRdd()
newDataRdd.foreachPartition(iter => {
val partitionId: Int = TaskContext.get.partitionId
val dictionaryMap: ConcurrentHashMap[String, T] = hisBc.value.get(partitionId)
val joinRdd = getJoinRdd()
val fqdnAccmu = SparkSessionUtil.getLongAccumulator(s"$collName Accumulator")
joinRdd.foreachPartition(iter => {
val resultDocumentList = new util.ArrayList[T]
var i = 0
iter.foreach(row => {
val document = getDocumentRow(row, dictionaryMap)
resultDocumentList.add(document)
val document: T = getDocumentRow(row)
if (document != null) {
fqdnAccmu.add(1)
// println(document)
resultDocumentList.add(document)
}
i += 1
if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
arangoManger.overwrite(resultDocumentList, collName)
@@ -69,92 +87,229 @@ object UpdateDocument {
LOG.warn(s"更新$collName:" + i)
}
})
LOG.warn(s"更新$collName 条数:${fqdnAccmu.value}")
val last = System.currentTimeMillis()
LOG.warn(s"更新$collName 时间${last - start}")
LOG.warn(s"更新$collName 时间:${last - start}")
} catch {
case e: Exception => e.printStackTrace()
} finally {
hisBc.destroy()
}
}
private def getVertexFqdnRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseDocument]): BaseDocument = {
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 = dictionaryMap.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)
}
private def insertFrameIp(): Unit = {
mergeVertexFrameIp.foreachPartition(iter => {
val resultDocumentList = new util.ArrayList[BaseDocument]
var i = 0
iter.foreach(row => {
val document = getVertexFrameipRow(row)
resultDocumentList.add(document)
i += 1
if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
arangoManger.overwrite(resultDocumentList, "IP")
LOG.warn(s"更新:IP" + i)
i = 0
}
})
if (i != 0) {
arangoManger.overwrite(resultDocumentList, "IP")
LOG.warn(s"更新IP:" + i)
}
})
}
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 + "-" + vsysId)
document.addAttribute("IP", ip)
document
}
private def getVertexIpRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseDocument]): BaseDocument = {
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)
private def getRelationSubidLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Row))): BaseEdgeDocument = {
var document = dictionaryMap.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", "")
val subidLocIpDocOpt = joinRow._2._1
var subidLocIpDoc = subidLocIpDocOpt match {
case Some(doc) => doc
case None => null
}
document
val subidLocIpRow = joinRow._2._2
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 + "-" + 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 + "-" + 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
}
private def getRelationFqdnLocateIpRow(row: Row, dictionaryMap: ConcurrentHashMap[String, BaseEdgeDocument]): BaseEdgeDocument = {
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 = dictionaryMap.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)
private def getVertexSubidRow(joinRow: (String, (Option[BaseDocument], Row))): BaseDocument = {
val subidDocOpt = joinRow._2._1
var subidDoc = subidDocOpt match {
case Some(doc) => doc
case None => null
}
document
val subidRow = joinRow._2._2
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(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
if (fqdnRow != null) {
val fqdn = fqdnRow.getAs[String]("FQDN")
val lastFoundTime = fqdnRow.getAs[Long]("LAST_FOUND_TIME")
val firstFoundTime = fqdnRow.getAs[Long]("FIRST_FOUND_TIME")
val vsysId = fqdnRow.getAs[Long]("VSYS_ID")
if (fqdnDoc != null) {
updateMaxAttribute(fqdnDoc, lastFoundTime, "LAST_FOUND_TIME")
fqdnDoc.addAttribute("VSYS_ID", vsysId)
} else {
fqdnDoc = new BaseDocument
fqdnDoc.setKey(fqdn + "-" + vsysId)
fqdnDoc.addAttribute("FQDN_NAME", fqdn)
fqdnDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
fqdnDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
fqdnDoc.addAttribute("VSYS_ID", vsysId)
}
}
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
if (ipRow != null) {
val ip = ipRow.getAs[String]("IP")
val firstFoundTime = ipRow.getAs[Long]("FIRST_FOUND_TIME")
val lastFoundTime = ipRow.getAs[Long]("LAST_FOUND_TIME")
val sessionCountList = ipRow.getAs[ofRef[AnyRef]]("SESSION_COUNT_LIST")
val bytesSumList = ipRow.getAs[ofRef[AnyRef]]("BYTES_SUM_LIST")
val ipTypeList = ipRow.getAs[ofRef[String]]("ip_type_list")
val linkInfo = ipRow.getAs[String]("common_link_info")
val sepAttributeTuple = separateAttributeByIpType(ipTypeList, sessionCountList, bytesSumList)
val vsysId = ipRow.getAs[Long]("VSYS_ID")
if (ipDoc != null) {
updateMaxAttribute(ipDoc, lastFoundTime, "LAST_FOUND_TIME")
updateSumAttribute(ipDoc, sepAttributeTuple._1, "SERVER_SESSION_COUNT")
updateSumAttribute(ipDoc, sepAttributeTuple._2, "SERVER_BYTES_SUM")
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 + "-" + vsysId)
ipDoc.addAttribute("IP", ip)
ipDoc.addAttribute("FIRST_FOUND_TIME", firstFoundTime)
ipDoc.addAttribute("LAST_FOUND_TIME", lastFoundTime)
ipDoc.addAttribute("SERVER_SESSION_COUNT", sepAttributeTuple._1)
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", linkInfo)
ipDoc.addAttribute("VSYS_ID", vsysId)
}
}
ipDoc
}
private def getRelationFqdnLocateIpRow(joinRow: (String, (Option[BaseEdgeDocument], Row))): BaseEdgeDocument = {
val fqdnLocIpDocOpt = joinRow._2._1
var fqdnLocIpDoc = fqdnLocIpDocOpt match {
case Some(doc) => doc
case None => null
}
val fqdnLocIpRow = joinRow._2._2
if (fqdnLocIpDoc != null) {
updateProtocolDocument(fqdnLocIpDoc)
}
if (fqdnLocIpRow != null) {
val fqdn = fqdnLocIpRow.getAs[String]("FQDN")
val serverIp = fqdnLocIpRow.getAs[String]("common_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 distCipRecent = fqdnLocIpRow.getAs[ofRef[String]]("DIST_CIP_RECENT")
val vsysId = fqdnLocIpRow.getAs[Long]("VSYS_ID")
val sepAttritubeMap: Map[String, Long] = separateAttributeByProtocol(schemaTypeList, countTotalList)
val distinctIp: Array[String] = mergeDistinctIp(distCipRecent)
val key = fqdn.concat("-" + serverIp + "-" + 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+ "-" + 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
}
}

View File

@@ -0,0 +1,136 @@
/*
* DISCLAIMER
*
* Copyright 2016 ArangoDB GmbH, Cologne, Germany
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Copyright holder is ArangoDB GmbH, Cologne, Germany
*
* author Mark - mark at arangodb.com
*/
package cn.ac.iie.spark
import cn.ac.iie.spark.rdd.{ArangoRdd, ReadOptions, WriteOptions}
import cn.ac.iie.spark.vpack.VPackUtils
import com.arangodb.model.DocumentCreateOptions
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import scala.collection.JavaConverters.seqAsJavaListConverter
import scala.reflect.ClassTag
object ArangoSpark {
/**
* Save data from rdd into ArangoDB
*
* @param rdd the rdd with the data to save
* @param collection the collection to save in
*/
def save[T](rdd: RDD[T], collection: String): Unit =
save(rdd, collection, WriteOptions())
/**
* Save data from rdd into ArangoDB
*
* @param rdd the rdd with the data to save
* @param collection the collection to save in
* @param options additional write options
*/
def save[T](rdd: RDD[T], collection: String, options: WriteOptions): Unit =
saveRDD(rdd, collection, options, (x: Iterator[T]) => x)
/**
* Save data from dataset into ArangoDB
*
* @param dataset the dataset with data to save
* @param collection the collection to save in
*/
def save[T](dataset: Dataset[T], collection: String): Unit =
saveRDD(dataset.rdd, collection, WriteOptions(), (x: Iterator[T]) => x)
/**
* Save data from dataset into ArangoDB
*
* @param dataset the dataset with data to save
* @param collection the collection to save in
* @param options additional write options
*/
def save[T](dataset: Dataset[T], collection: String, options: WriteOptions): Unit =
saveRDD(dataset.rdd, collection, options, (x: Iterator[T]) => x)
/**
* Save data from dataframe into ArangoDB
*
* @param dataframe the dataframe with data to save
* @param collection the collection to save in
*/
def saveDF(dataframe: DataFrame, collection: String): Unit =
saveRDD[Row](dataframe.rdd, collection, WriteOptions(), (x: Iterator[Row]) => x.map { y => VPackUtils.rowToVPack(y) })
/**
* Save data from dataframe into ArangoDB
*
* @param dataframe the dataframe with data to save
* @param collection the collection to save in
* @param options additional write options
*/
def saveDF(dataframe: DataFrame, collection: String, options: WriteOptions): Unit =
saveRDD[Row](dataframe.rdd, collection, options, (x: Iterator[Row]) => x.map { y => VPackUtils.rowToVPack(y) })
private def saveRDD[T](rdd: RDD[T], collection: String, options: WriteOptions, map: Iterator[T] => Iterator[Any]): Unit = {
val writeOptions = createWriteOptions(options, rdd.sparkContext.getConf)
rdd.foreachPartition { p =>
if (p.nonEmpty) {
val arangoDB = createArangoBuilder(writeOptions).build()
val col = arangoDB.db(writeOptions.database).collection(collection)
val docs = map(p).toList.asJava
writeOptions.method match {
case WriteOptions.INSERT => col.insertDocuments(docs)
case WriteOptions.UPDATE => col.updateDocuments(docs)
case WriteOptions.REPLACE => col.replaceDocuments(docs)
case WriteOptions.OVERWRITE =>
val documentCreateOptions = new DocumentCreateOptions
documentCreateOptions.overwrite(true)
documentCreateOptions.silent(true)
col.insertDocuments(docs, documentCreateOptions)
}
arangoDB.shutdown()
}
}
}
/**
* Load data from ArangoDB into rdd
*
* @param sparkContext the sparkContext containing the ArangoDB configuration
* @param collection the collection to load data from
*/
def load[T: ClassTag](sparkContext: SparkContext, collection: String): ArangoRdd[T] =
load(sparkContext, collection, ReadOptions())
/**
* Load data from ArangoDB into rdd
*
* @param sparkContext the sparkContext containing the ArangoDB configuration
* @param collection the collection to load data from
* @param options read options
*/
def load[T: ClassTag](sparkContext: SparkContext, collection: String, options: ReadOptions): ArangoRdd[T] =
new ArangoRdd[T](sparkContext, createReadOptions(options, sparkContext.getConf).copy(collection = collection))
}

View File

@@ -0,0 +1,7 @@
package cn.ac.iie.spark.partition
import org.apache.spark.Partition
class QueryArangoPartition(idx: Int, val offset: Long, val separate: Long) extends Partition{
override def index: Int = idx
}

View File

@@ -0,0 +1,34 @@
package cn.ac.iie.spark.rdd
import com.arangodb.Protocol
import com.arangodb.entity.LoadBalancingStrategy
trait ArangoOptions {
def database: String = "_system"
def hosts: Option[String] = None
def user: Option[String] = None
def password: Option[String] = None
def useSsl: Option[Boolean] = None
def sslKeyStoreFile: Option[String] = None
def sslPassPhrase: Option[String] = None
def sslProtocol: Option[String] = None
def protocol: Option[Protocol] = None
def maxConnections: Option[Int] = None
def acquireHostList: Option[Boolean] = None
def acquireHostListInterval: Option[Int] = None
def loadBalancingStrategy: Option[LoadBalancingStrategy] = None
}

View File

@@ -0,0 +1,90 @@
package cn.ac.iie.spark.rdd
import java.util
import scala.collection.JavaConverters.asScalaIteratorConverter
import cn.ac.iie.config.ApplicationConfig
import cn.ac.iie.service.update.UpdateDocument
import cn.ac.iie.spark
import cn.ac.iie.spark.partition.QueryArangoPartition
import com.arangodb.ArangoCursor
import com.arangodb.model.AqlQueryOptions
import com.arangodb.util.MapBuilder
import org.apache.spark.{Partition, SparkContext, TaskContext}
import org.apache.spark.rdd.RDD
import org.slf4j.LoggerFactory
import scala.collection.mutable.ArrayBuffer
import scala.reflect.ClassTag
class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
val options: ReadOptions
) extends RDD[T](sparkContext, Nil) {
private val LOG = LoggerFactory.getLogger(UpdateDocument.getClass)
override def compute(split: Partition, context: TaskContext): Iterator[T] = {
createCursor(split.asInstanceOf[QueryArangoPartition]).asScala
}
override protected def getPartitions: Array[Partition] = {
val partitions = ArrayBuffer[Partition]()
val total = getCountTotal
for (i <- 0 until ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS) {
val partition = getPartition(i, total)
partitions += partition
}
partitions.toArray
}
private def createCursor(split: QueryArangoPartition)(implicit clazz: ClassTag[T]): ArangoCursor[T] = {
var arangoCursor:ArangoCursor[T] = null
val arangoDB = spark.createArangoBuilder(options).build()
val bindVars: util.Map[String, AnyRef] = new MapBuilder().get
val queryOptions: AqlQueryOptions = new AqlQueryOptions().ttl(ApplicationConfig.ARANGODB_TTL)
try {
val offset = split.offset
val separate = split.separate
val collection = options.collection
val sql = s"FOR doc IN $collection limit $offset,$separate RETURN doc"
LOG.info(sql)
arangoCursor = arangoDB.db(options.database).query(sql,bindVars,queryOptions,clazz.runtimeClass.asInstanceOf[Class[T]])
}catch {
case e: Exception => LOG.error(s"创建Cursor异常:${e.getMessage}")
}finally {
arangoDB.shutdown()
}
arangoCursor
}
override def repartition(numPartitions: Int)(implicit ord: Ordering[T]): RDD[T] = super.repartition(numPartitions)
private def getPartition(idx: Int, countTotal: Long): QueryArangoPartition = {
val sepNum = countTotal / ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS + 1
val offsetNum = idx * sepNum
new QueryArangoPartition(idx, offsetNum, sepNum)
}
override def count(): Long = getCountTotal
private def getCountTotal: Long = {
val arangoDB = spark.createArangoBuilder(options).build()
var cnt = 0L
val sql = s"RETURN LENGTH(${options.collection})"
LOG.info(sql)
try {
val longs = arangoDB.db(options.database).query(sql, classOf[Long])
while (longs.hasNext) cnt = longs.next
if (cnt > ApplicationConfig.ARANGODB_TOTAL_NUM){
cnt = ApplicationConfig.ARANGODB_TOTAL_NUM
}
} catch {
case e: Exception => LOG.error(sql + s"执行异常:${e.getMessage}")
}finally {
arangoDB.shutdown()
}
cnt
}
}

View File

@@ -0,0 +1,93 @@
/*
* DISCLAIMER
*
* Copyright 2016 ArangoDB GmbH, Cologne, Germany
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Copyright holder is ArangoDB GmbH, Cologne, Germany
*
* author Mark - mark at arangodb.com
*/
package cn.ac.iie.spark.rdd
import cn.ac.iie.spark.partition.QueryArangoPartition
import com.arangodb.Protocol
import com.arangodb.entity.LoadBalancingStrategy
case class ReadOptions(override val database: String = "_system",
val collection: String = null,
partitioner: QueryArangoPartition = new QueryArangoPartition(0,0,0),
override val hosts: Option[String] = None,
override val user: Option[String] = None,
override val password: Option[String] = None,
override val useSsl: Option[Boolean] = None,
override val sslKeyStoreFile: Option[String] = None,
override val sslPassPhrase: Option[String] = None,
override val sslProtocol: Option[String] = None,
override val protocol: Option[Protocol] = None,
override val maxConnections: Option[Int] = None,
override val acquireHostList: Option[Boolean] = None,
override val acquireHostListInterval: Option[Int] = None,
override val loadBalancingStrategy: Option[LoadBalancingStrategy] = None) extends ArangoOptions {
def this() = this(database = "_system")
def database(database: String): ReadOptions = copy(database = database)
def collection(collection: String): ReadOptions = copy(collection = collection)
def hosts(hosts: String): ReadOptions = copy(hosts = Some(hosts))
def user(user: String): ReadOptions = copy(user = Some(user))
def password(password: String): ReadOptions = copy(password = Some(password))
def useSsl(useSsl: Boolean): ReadOptions = copy(useSsl = Some(useSsl))
def sslKeyStoreFile(sslKeyStoreFile: String): ReadOptions = copy(sslKeyStoreFile = Some(sslKeyStoreFile))
def sslPassPhrase(sslPassPhrase: String): ReadOptions = copy(sslPassPhrase = Some(sslPassPhrase))
def sslProtocol(sslProtocol: String): ReadOptions = copy(sslProtocol = Some(sslProtocol))
def protocol(protocol: Protocol): ReadOptions = copy(protocol = Some(protocol))
def maxConnections(maxConnections: Int): ReadOptions = copy(maxConnections = Some(maxConnections))
def acquireHostList(acquireHostList: Boolean): ReadOptions = copy(acquireHostList = Some(acquireHostList))
def acquireHostListInterval(acquireHostListInterval: Int): ReadOptions = copy(acquireHostListInterval = Some(acquireHostListInterval))
def loadBalancingStrategy(loadBalancingStrategy: LoadBalancingStrategy): ReadOptions = copy(loadBalancingStrategy = Some(loadBalancingStrategy))
def copy(database: String = database,
collection: String = collection,
partitioner: QueryArangoPartition = partitioner,
hosts: Option[String] = hosts,
user: Option[String] = user,
password: Option[String] = password,
useSsl: Option[Boolean] = useSsl,
sslKeyStoreFile: Option[String] = sslKeyStoreFile,
sslPassPhrase: Option[String] = sslPassPhrase,
sslProtocol: Option[String] = sslProtocol,
protocol: Option[Protocol] = protocol,
maxConnections: Option[Int] = maxConnections,
acquireHostList: Option[Boolean] = acquireHostList,
acquireHostListInterval: Option[Int] = acquireHostListInterval,
loadBalancingStrategy: Option[LoadBalancingStrategy] = loadBalancingStrategy): ReadOptions = {
ReadOptions(database, collection, partitioner, hosts, user, password, useSsl, sslKeyStoreFile, sslPassPhrase, sslProtocol, protocol, maxConnections, acquireHostList, acquireHostListInterval, loadBalancingStrategy)
}
}

View File

@@ -0,0 +1,124 @@
/*
* DISCLAIMER
*
* Copyright 2016 ArangoDB GmbH, Cologne, Germany
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Copyright holder is ArangoDB GmbH, Cologne, Germany
*
* author Mark - mark at arangodb.com
*/
package cn.ac.iie.spark.rdd
import javax.net.ssl.SSLContext
import com.arangodb.Protocol
import com.arangodb.entity.LoadBalancingStrategy
case class WriteOptions(override val database: String = "_system",
val method: WriteOptions.Method = WriteOptions.INSERT,
override val hosts: Option[String] = None,
override val user: Option[String] = None,
override val password: Option[String] = None,
override val useSsl: Option[Boolean] = None,
override val sslKeyStoreFile: Option[String] = None,
override val sslPassPhrase: Option[String] = None,
override val sslProtocol: Option[String] = None,
override val protocol: Option[Protocol] = None,
override val maxConnections: Option[Int] = None,
override val acquireHostList: Option[Boolean] = None,
override val acquireHostListInterval: Option[Int] = None,
override val loadBalancingStrategy: Option[LoadBalancingStrategy] = None) extends ArangoOptions {
import WriteOptions._
def this() = this(database = "_system")
def database(database: String): WriteOptions = copy(database = database)
def method(method: Method): WriteOptions = copy(method = method)
def hosts(hosts: String): WriteOptions = copy(hosts = Some(hosts))
def user(user: String): WriteOptions = copy(user = Some(user))
def password(password: String): WriteOptions = copy(password = Some(password))
def useSsl(useSsl: Boolean): WriteOptions = copy(useSsl = Some(useSsl))
def sslKeyStoreFile(sslKeyStoreFile: String): WriteOptions = copy(sslKeyStoreFile = Some(sslKeyStoreFile))
def sslPassPhrase(sslPassPhrase: String): WriteOptions = copy(sslPassPhrase = Some(sslPassPhrase))
def sslProtocol(sslProtocol: String): WriteOptions = copy(sslProtocol = Some(sslProtocol))
def protocol(protocol: Protocol): WriteOptions = copy(protocol = Some(protocol))
def maxConnections(maxConnections: Int): WriteOptions = copy(maxConnections = Some(maxConnections))
def acquireHostList(acquireHostList: Boolean): WriteOptions = copy(acquireHostList = Some(acquireHostList))
def acquireHostListInterval(acquireHostListInterval: Int): WriteOptions = copy(acquireHostListInterval = Some(acquireHostListInterval))
def loadBalancingStrategy(loadBalancingStrategy: LoadBalancingStrategy): WriteOptions = copy(loadBalancingStrategy = Some(loadBalancingStrategy))
def copy(database: String = database,
method: Method = method,
hosts: Option[String] = hosts,
user: Option[String] = user,
password: Option[String] = password,
useSsl: Option[Boolean] = useSsl,
sslKeyStoreFile: Option[String] = sslKeyStoreFile,
sslPassPhrase: Option[String] = sslPassPhrase,
sslProtocol: Option[String] = sslProtocol,
protocol: Option[Protocol] = protocol,
maxConnections: Option[Int] = maxConnections,
acquireHostList: Option[Boolean] = acquireHostList,
acquireHostListInterval: Option[Int] = acquireHostListInterval,
loadBalancingStrategy: Option[LoadBalancingStrategy] = loadBalancingStrategy): WriteOptions = {
WriteOptions(database, method, hosts, user, password, useSsl, sslKeyStoreFile, sslPassPhrase, sslProtocol, protocol, maxConnections, acquireHostList, acquireHostListInterval, loadBalancingStrategy)
}
}
object WriteOptions {
/**
* method to save documents to arangodb
*/
sealed trait Method
/**
* save documents by inserting
* @see [[com.arangodb.ArangoCollection#insertDocuments(java.util.Collection)]]
*/
case object INSERT extends Method
/**
* save documents by updating
* @see [[com.arangodb.ArangoCollection#updateDocuments(java.util.Collection)]]
*/
case object UPDATE extends Method
/**
* save documents by replacing
* @see [[com.arangodb.ArangoCollection#replaceDocuments(java.util.Collection)]]
*/
case object REPLACE extends Method
/**
* save documents by overwrite
*/
case object OVERWRITE extends Method
}

View File

@@ -0,0 +1,139 @@
/*
* DISCLAIMER
*
* Copyright 2016 ArangoDB GmbH, Cologne, Germany
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Copyright holder is ArangoDB GmbH, Cologne, Germany
*
* author Mark - mark at arangodb.com
*/
package cn.ac.iie
import java.io.FileInputStream
import java.security.KeyStore
import cn.ac.iie.spark.rdd.{ArangoOptions, ReadOptions, WriteOptions}
import com.arangodb.{ArangoDB, ArangoDBException, Protocol}
import com.arangodb.entity.LoadBalancingStrategy
import com.arangodb.velocypack.module.jdk8.VPackJdk8Module
import com.arangodb.velocypack.module.scala.VPackScalaModule
import javax.net.ssl.{KeyManagerFactory, SSLContext, TrustManagerFactory}
import org.apache.spark.SparkConf
import scala.util.Try
package object spark {
val PropertyHosts = "arangodb.hosts"
val PropertyUser = "arangodb.user"
val PropertyPassword = "arangodb.password"
val PropertyUseSsl = "arangodb.useSsl"
val PropertySslKeyStoreFile = "arangodb.ssl.keyStoreFile"
val PropertySslPassPhrase = "arangodb.ssl.passPhrase"
val PropertySslProtocol = "arangodb.ssl.protocol"
val PropertyProtocol = "arangodb.protocol"
val PropertyMaxConnections = "arangodb.maxConnections"
val PropertyAcquireHostList = "arangodb.acquireHostList"
val PropertyAcquireHostListInterval = "arangodb.acquireHostListInterval"
val PropertyLoadBalancingStrategy = "arangodb.loadBalancingStrategy"
private[spark] def createReadOptions(options: ReadOptions, sc: SparkConf): ReadOptions = {
options.copy(
hosts = options.hosts.orElse(some(sc.get(PropertyHosts, null))),
user = options.user.orElse(some(sc.get(PropertyUser, null))),
password = options.password.orElse(some(sc.get(PropertyPassword, null))),
useSsl = options.useSsl.orElse(some(Try(sc.get(PropertyUseSsl, null).toBoolean).getOrElse(false))),
sslKeyStoreFile = options.sslKeyStoreFile.orElse(some(sc.get(PropertySslKeyStoreFile, null))),
sslPassPhrase = options.sslPassPhrase.orElse(some(sc.get(PropertySslPassPhrase, null))),
sslProtocol = options.sslProtocol.orElse(some(sc.get(PropertySslProtocol, null))),
protocol = options.protocol.orElse(some(Protocol.valueOf(sc.get(PropertyProtocol, "VST")))),
maxConnections = options.maxConnections.orElse(some(Try(sc.get(PropertyMaxConnections, null).toInt).getOrElse(1))),
acquireHostList = options.acquireHostList.orElse(some(Try(sc.get(PropertyAcquireHostList, null).toBoolean).getOrElse(false))),
acquireHostListInterval = options.acquireHostListInterval.orElse(some(Try(sc.get(PropertyAcquireHostListInterval, null).toInt).getOrElse(60000))),
loadBalancingStrategy = options.loadBalancingStrategy.orElse(some(LoadBalancingStrategy.valueOf(sc.get(PropertyLoadBalancingStrategy, "NONE")))))
}
private[spark] def createWriteOptions(options: WriteOptions, sc: SparkConf): WriteOptions = {
options.copy(
hosts = options.hosts.orElse(some(sc.get(PropertyHosts, null))),
user = options.user.orElse(some(sc.get(PropertyUser, null))),
password = options.password.orElse(some(sc.get(PropertyPassword, null))),
useSsl = options.useSsl.orElse(some(Try(sc.get(PropertyUseSsl, null).toBoolean).getOrElse(false))),
sslKeyStoreFile = options.sslKeyStoreFile.orElse(some(sc.get(PropertySslKeyStoreFile, null))),
sslPassPhrase = options.sslPassPhrase.orElse(some(sc.get(PropertySslPassPhrase, null))),
sslProtocol = options.sslProtocol.orElse(some(sc.get(PropertySslProtocol, null))),
protocol = options.protocol.orElse(some(Protocol.valueOf(sc.get(PropertyProtocol, "VST")))),
maxConnections = options.maxConnections.orElse(some(Try(sc.get(PropertyMaxConnections, null).toInt).getOrElse(1))),
acquireHostList = options.acquireHostList.orElse(some(Try(sc.get(PropertyAcquireHostList, null).toBoolean).getOrElse(false))),
acquireHostListInterval = options.acquireHostListInterval.orElse(some(Try(sc.get(PropertyAcquireHostListInterval, null).toInt).getOrElse(60000))),
loadBalancingStrategy = options.loadBalancingStrategy.orElse(some(LoadBalancingStrategy.valueOf(sc.get(PropertyLoadBalancingStrategy, "NONE")))))
}
private[spark] def createArangoBuilder(options: ArangoOptions): ArangoDB.Builder = {
val builder = new ArangoDB.Builder()
builder.registerModules(new VPackJdk8Module, new VPackScalaModule)
options.hosts.foreach { hosts(_).foreach(host => builder.host(host._1, host._2)) }
options.user.foreach { builder.user(_) }
options.password.foreach { builder.password(_) }
options.useSsl.foreach { builder.useSsl(_) }
if (options.sslKeyStoreFile.isDefined && options.sslPassPhrase.isDefined) {
builder.sslContext(createSslContext(options.sslKeyStoreFile.get, options.sslPassPhrase.get, options.sslProtocol.getOrElse("TLS")))
}
options.protocol.foreach { builder.useProtocol(_) }
options.maxConnections.foreach { builder.maxConnections(_) }
options.acquireHostList.foreach { builder.acquireHostList(_) }
options.acquireHostListInterval.foreach { builder.acquireHostListInterval(_) }
options.loadBalancingStrategy.foreach { builder.loadBalancingStrategy(_) }
builder
}
private def createSslContext(keyStoreFile: String, passPhrase: String, protocol: String): SSLContext = {
val ks = KeyStore.getInstance(KeyStore.getDefaultType());
val kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
ks.load(new FileInputStream(keyStoreFile), passPhrase.toCharArray());
kmf.init(ks, passPhrase.toCharArray());
val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ks);
val sc = SSLContext.getInstance(protocol);
sc.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
sc
}
private def some(value: String): Option[String] =
if (value != null) Some(value) else None
private def some(value: Int): Option[Int] =
Some(value)
private def some(value: Boolean): Option[Boolean] =
Some(value)
private def some(value: Protocol): Option[Protocol] =
Some(value)
private def some(value: LoadBalancingStrategy): Option[LoadBalancingStrategy] =
Some(value)
private def hosts(hosts: String): List[(String, Int)] =
hosts.split(",").map({ x =>
val s = x.split(":")
if (s.length != 2 || !s(1).matches("[0-9]+"))
throw new ArangoDBException(s"Could not load property-value arangodb.hosts=${s}. Expected format ip:port,ip:port,...");
else
(s(0), s(1).toInt)
}).toList
}

View File

@@ -0,0 +1,160 @@
/*
* DISCLAIMER
*
* Copyright 2016 ArangoDB GmbH, Cologne, Germany
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* Copyright holder is ArangoDB GmbH, Cologne, Germany
*
* author Mark - mark at arangodb.com
*/
package cn.ac.iie.spark.vpack
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{
ArrayType,
BooleanType,
DataType,
DateType,
DecimalType,
DoubleType,
FloatType,
IntegerType,
LongType,
MapType,
NullType,
ShortType,
StringType,
StructField,
StructType,
TimestampType
}
import com.arangodb.velocypack.VPackBuilder
import com.arangodb.velocypack.VPackSlice
import com.arangodb.velocypack.ValueType
private[spark] object VPackUtils {
def rowToVPack(row: Row): VPackSlice = {
val builder = new VPackBuilder()
if (row == null) {
builder.add(ValueType.NULL)
} else {
builder.add(ValueType.OBJECT)
row.schema.fields.zipWithIndex.foreach { addField(_, row, builder) }
builder.close()
}
builder.slice()
}
private def addField(field: (StructField, Int), row: Row, builder: VPackBuilder): Unit = {
val name = field._1.name
val index = field._2
if (row.isNullAt(index)) {
builder.add(name, ValueType.NULL)
} else {
field._1.dataType match {
case BooleanType => builder.add(name, java.lang.Boolean.valueOf(row.getBoolean(index)))
case DoubleType => builder.add(name, java.lang.Double.valueOf(row.getDouble(index)))
case FloatType => builder.add(name, java.lang.Float.valueOf(row.getFloat(index)))
case LongType => builder.add(name, java.lang.Long.valueOf(row.getLong(index)))
case IntegerType => builder.add(name, java.lang.Integer.valueOf(row.getInt(index)))
case ShortType => builder.add(name, java.lang.Short.valueOf(row.getShort(index)))
case StringType => builder.add(name, java.lang.String.valueOf(row.getString(index)));
case DateType => builder.add(name, row.getDate(index))
case TimestampType => builder.add(name, row.getTimestamp(index))
case t: DecimalType => builder.add(name, row.getDecimal(index))
case t: MapType => {
builder.add(name, ValueType.OBJECT)
row.getMap[String, Any](index).foreach { case (name, value) => addValue(name, value, builder) }
builder.close()
}
case t: ArrayType => {
builder.add(name, ValueType.ARRAY)
addValues(row, index, builder, t.elementType)
builder.close()
}
case NullType => builder.add(name, ValueType.NULL)
case struct: StructType => builder.add(name, rowToVPack(row.getStruct(index)))
case _ => // TODO
}
}
}
private def addValues(row: Row, index: Int, builder: VPackBuilder, itemType: DataType): Unit = {
itemType match {
case BooleanType =>
row.getSeq[Boolean](index).foreach { value =>
addValue(null, value, builder)
}
case DoubleType =>
row.getSeq[Double](index).foreach { value =>
addValue(null, value, builder)
}
case FloatType =>
row.getSeq[Float](index).foreach { value =>
addValue(null, value, builder)
}
case LongType =>
row.getSeq[Long](index).foreach { value =>
addValue(null, value, builder)
}
case IntegerType =>
row.getSeq[Int](index).foreach { value =>
addValue(null, value, builder)
}
case ShortType =>
row.getSeq[Short](index).foreach { value =>
addValue(null, value, builder)
}
case StringType =>
row.getSeq[String](index).foreach { value =>
addValue(null, value, builder)
}
case DateType =>
row.getSeq[java.sql.Date](index).foreach { value =>
addValue(null, value, builder)
}
case TimestampType =>
row.getSeq[java.sql.Timestamp](index).foreach { value =>
addValue(null, value, builder)
}
case s: StructType => {
row.getSeq[Row](index).foreach { value =>
builder.add(null, rowToVPack(value))
}
}
case t: MapType => // TODO
case t: ArrayType => // TODO
case _ => // TODO
}
}
private def addValue(name: String, value: Any, builder: VPackBuilder): Unit = {
value match {
case value: Boolean => builder.add(name, java.lang.Boolean.valueOf(value))
case value: Double => builder.add(name, java.lang.Double.valueOf(value))
case value: Float => builder.add(name, java.lang.Float.valueOf(value))
case value: Long => builder.add(name, java.lang.Long.valueOf(value))
case value: Int => builder.add(name, java.lang.Integer.valueOf(value))
case value: Short => builder.add(name, java.lang.Short.valueOf(value))
case value: String => builder.add(name, java.lang.String.valueOf(value))
case value: java.sql.Date => builder.add(name, value)
case value: java.sql.Timestamp => builder.add(name, value)
case _ => // TODO
}
}
}

View File

@@ -1,7 +1,9 @@
package cn.ac.iie.utils
import cn.ac.iie.config.ApplicationConfig
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.util.LongAccumulator
import org.slf4j.LoggerFactory
object SparkSessionUtil {
@@ -9,6 +11,8 @@ object SparkSessionUtil {
val spark: SparkSession = getSparkSession
var sparkContext: SparkContext = getContext
private def getSparkSession: SparkSession ={
val spark: SparkSession = SparkSession
.builder()
@@ -17,16 +21,38 @@ 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)
.master(ApplicationConfig.MASTER)
.getOrCreate()
LOG.warn("sparkession获取成功")
spark
}
private def getContext: SparkContext = {
@transient var sc: SparkContext = null
if (sparkContext == null) sc = spark.sparkContext
sc
}
def getLongAccumulator(name: String): LongAccumulator ={
if (sparkContext == null){
sparkContext = getContext
}
sparkContext.longAccumulator(name)
}
def closeSpark(): Unit ={
if (spark != null){
spark.stop()
}
if (sparkContext != null){
sparkContext.stop()
}
}
}

View File

@@ -7,7 +7,7 @@ import org.apache.spark.sql.SparkSession
object BaseClickhouseDataTest {
private val spark: SparkSession = SparkSessionUtil.spark
def main(args: Array[String]): Unit = {
BaseClickhouseData loadConnectionDataFromCk()
// BaseClickhouseData loadConnectionDataFromCk()
val sql =
"""
|SELECT

View File

@@ -1,35 +0,0 @@
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)
}
}
}
}

View File

@@ -0,0 +1,75 @@
package cn.ac.iie.spark
import cn.ac.iie.config.ApplicationConfig
import cn.ac.iie.dao.BaseClickhouseData
import cn.ac.iie.spark.partition.CustomPartitioner
import cn.ac.iie.spark.rdd.{ArangoRdd, ReadOptions}
import cn.ac.iie.utils.SparkSessionUtil
import com.arangodb.entity.BaseDocument
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.{collect_list, max, min}
import org.apache.spark.storage.StorageLevel
object RDDTest {
def main(args: Array[String]): Unit = {
val sparkContext = SparkSessionUtil.spark.sparkContext
println(sparkContext.getConf.get("arangodb.hosts"))
// val options = ReadOptions("iplearn_media_domain").copy(collection = "R_LOCATE_FQDN2IP")
val options = ReadOptions(ApplicationConfig.ARANGODB_DB_NAME)
val ipOptions = options.copy(collection = "IP")
val rdd: ArangoRdd[BaseDocument] = ArangoSpark.load[BaseDocument](sparkContext,"IP",options)
println(rdd.count())
println(rdd.getNumPartitions)
val ipRDD = mergeVertexIp()
val value: RDD[(String, (Option[BaseDocument], Option[Row]))] = rdd.map(doc => {
(doc.getKey, doc)
}).fullOuterJoin(ipRDD)
value.foreach((row: (String, (Option[BaseDocument], Option[Row]))) => {
val value = row._2._2
val str: String = value match {
case Some(r) => r.getAs[String]("IP")
// case None => null
case _ => null
}
println(str)
})
/*
val value: RDD[BaseDocument] = rdd.filter(doc => doc.getAttribute("CLIENT_SESSION_COUNT").asInstanceOf[Long] > 100).map(doc => {
doc.addAttribute("abc", 1)
doc
})
value.map(doc => {(doc.getKey,doc)})
value.persist(StorageLevel.MEMORY_AND_DISK)
value.foreach(fqdnRow => println(fqdnRow.toString))
println(value.count())
*/
SparkSessionUtil.spark.close()
System.exit(0)
}
def mergeVertexIp(): RDD[(String,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.getAs[String]("IP"), row))
.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))
values
}
}