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

[修复] 增加iplearning对IP和VSYS_ID的非空判断，修改日志信息中文为英文

Browse Source
This commit is contained in:
zhanghongqing
2024-11-25 15:43:18 +08:00
parent 5ed7657f1e
commit 88451c2b3a
6 changed files with 31 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ip-learning-spark/src/main/java/cn/ac/iie/utils/ArangoDBConnect.java
View File

@@ -83,8 +83,7 @@ public class ArangoDBConnect {
collection.replaceDocuments(docUpdate);
}
} catch (Exception e) {
System.out.println("更新失败");
e.printStackTrace();
LOG.error("update failure：" + e.toString());
} finally {
docInsert.clear();
docInsert.clear();
@@ -102,11 +101,11 @@ public class ArangoDBConnect {
MultiDocumentEntity<DocumentCreateEntity<T>> documentCreateEntityMultiDocumentEntity = collection.insertDocuments(docOverwrite, documentCreateOptions);
Collection<ErrorEntity> errors = documentCreateEntityMultiDocumentEntity.getErrors();
for (ErrorEntity errorEntity : errors) {
LOG.warn("写入arangoDB异常：" + errorEntity.getErrorMessage());
LOG.warn("write arangoDB error：" + errorEntity.getErrorMessage());
}
}
} catch (Exception e) {
LOG.error("更新失败：" + e.toString());
LOG.error("update failure：" + e.toString());
} finally {
docOverwrite.clear();
}

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

@@ -15,7 +15,7 @@ object BaseArangoData {
def loadArangoRdd[T: ClassTag](name:String): ArangoRdd[T] ={
val value = ArangoSpark.load[T](sparkContext, name, options)
LOG.warn(s"读取$name arangoDb:${value.count()}")
LOG.warn(s"read $name arangoDb: ${value.count()}")
value
}

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

@@ -1,13 +1,13 @@
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
import java.util.Date
object BaseClickhouseData {
private val LOG = LoggerFactory.getLogger(BaseClickhouseData.getClass)

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

@@ -50,7 +50,8 @@ object MergeDataFrame {
def mergeRelationFqdnLocateIp(): RDD[(String, (Option[BaseEdgeDocument], Row))] = {
val frame = BaseClickhouseData.getRelationFqdnLocateIpDf
.repartition().filter(row => isDomain(row.getAs[String]("FQDN")))
.repartition()
.filter(row => isDomain(row.getAs[String]("FQDN")))
.groupBy("FQDN", "server_ip", "VSYS_ID")
.agg(
min("FIRST_FOUND_TIME").alias("FIRST_FOUND_TIME"),
@@ -59,13 +60,21 @@ object MergeDataFrame {
collect_list("decoded_as_list").alias("decoded_as_list"),
collect_set("DIST_CIP_RECENT").alias("DIST_CIP_RECENT")
)
val fqdnLocIpRddRow = frame.rdd.map(row => {
val fqdn = row.getAs[String]("FQDN")
val serverIp = row.getAs[String]("server_ip")
val vsysId = row.getAs[Integer]("VSYS_ID").toLong
val key = fqdn.concat("-" + serverIp + "-" + vsysId)
(key, row)
}) /*.partitionBy(new CustomPartitioner(ApplicationConfig.SPARK_SQL_SHUFFLE_PARTITIONS))*/
val fqdnLocIpRddRow = frame.rdd
.filter(row => {
// 检查 server_ip 和 VSYS_ID 是否为 null
val serverIp = row.getAs[String]("server_ip")
val vsysId = row.getAs[Integer]("VSYS_ID")
serverIp != null && vsysId != null
})
.map(row => {
val fqdn = row.getAs[String]("FQDN")
val serverIp = row.getAs[String]("server_ip")
val vsysId = row.getAs[Integer]("VSYS_ID").toLong
val key = fqdn.concat("-" + serverIp + "-" + vsysId)
(key, row)
})
val fqdnLocIpRddDoc = BaseArangoData.loadArangoRdd[BaseEdgeDocument]("R_LOCATE_FQDN2IP")
fqdnLocIpRddDoc.map(doc => (doc.getKey, doc)).rightOuterJoin(fqdnLocIpRddRow)
@@ -133,7 +142,7 @@ object MergeDataFrame {
}
} catch {
case e: Exception =>
LOG.error("解析域名 " + fqdn + " 失败：\n" + e.toString)
LOG.error("Domain name resolution " + fqdn + " failure：\n" + e.toString)
}
false

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

@@ -78,20 +78,20 @@ object UpdateDocument {
i += 1
if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
arangoManger.overwrite(resultDocumentList, collName)
LOG.warn(s"更新:$collName" + i)
LOG.warn(s"update $collName: " + i)
i = 0
}
})
if (i != 0) {
arangoManger.overwrite(resultDocumentList, collName)
LOG.warn(s"更新$collName:" + i)
LOG.warn(s"update $collName: " + i)
}
})
LOG.warn(s"更新$collName 条数:${fqdnAccmu.value}")
LOG.warn(s"update $collName rows: ${fqdnAccmu.value}")
val last = System.currentTimeMillis()
LOG.warn(s"更新$collName 时间:${last - start}")
LOG.warn(s"update $collName time: ${last - start}")
} catch {
case e: Exception => e.printStackTrace()
}
@@ -107,13 +107,13 @@ object UpdateDocument {
i += 1
if (i >= ApplicationConfig.UPDATE_ARANGO_BATCH) {
arangoManger.overwrite(resultDocumentList, "IP")
LOG.warn(s"更新:IP" + i)
LOG.warn(s"update IP: " + i)
i = 0
}
})
if (i != 0) {
arangoManger.overwrite(resultDocumentList, "IP")
LOG.warn(s"更新IP:" + i)
LOG.warn(s"update IP: " + i)
}
})
}

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

@@ -29,7 +29,7 @@ object SparkSessionUtil {
.config("arangodb.password", ApplicationConfig.ARANGODB_PASSWORD)
.master(ApplicationConfig.MASTER)
.getOrCreate()
LOG.warn("sparkession获取成功！！！")
LOG.warn("spark session start success！！！")
spark
}