You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by ra...@apache.org on 2017/06/23 16:39:34 UTC
[04/50] [abbrv] incubator-spot git commit: flow_tuning
flow_tuning
reverted some changes to defer them for a larger overhaul
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/c0f80c56
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/c0f80c56
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/c0f80c56
Branch: refs/heads/SPOT-35_graphql_api
Commit: c0f80c56ea3f8cae666c3875fb87b3b175fdd915
Parents: 4004ede
Author: nlsegerl <na...@intel.com>
Authored: Tue Jun 6 10:26:20 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Tue Jun 6 10:26:20 2017 -0700
----------------------------------------------------------------------
.../FlowSuspiciousConnectsAnalysis.scala | 34 ++++++++++++++------
.../FlowSuspiciousConnectsAnalysisTest.scala | 4 +--
.../model/FlowSuspiciousConnectsModelTest.scala | 8 ++---
3 files changed, 30 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index e972bf4..7469e13 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -41,15 +41,11 @@ object FlowSuspiciousConnectsAnalysis {
logger.info("Starting flow suspicious connects analysis.")
- val flows : DataFrame = cleanFlowRecords(inputFlowRecords)
+ val flows : DataFrame = filterAndSelectCleanFlowRecords(inputFlowRecords)
- logger.info("Fitting probabilistic model to data")
- val model =
- FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows)
-
logger.info("Identifying outliers")
- val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
+ val scoredFlowRecords = detectFlowAnomalies(flows, config, sparkContext, sqlContext, logger)
val filteredFlowRecords = filterScoredFlowRecords(scoredFlowRecords, config.threshold)
@@ -71,6 +67,27 @@ object FlowSuspiciousConnectsAnalysis {
}
+ /**
+ * Identify anomalous netflow log entries in in the provided data frame.
+ *
+ * @param data Data frame of netflow entries
+ * @param config
+ * @param sparkContext
+ * @param sqlContext
+ * @param logger
+ * @return
+ */
+ def detectFlowAnomalies(data: DataFrame,
+ config: SuspiciousConnectsConfig,
+ sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger): DataFrame = {
+ logger.info("Fitting probabilistic model to data")
+ val model =
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, data)
+ logger.info("Identifying outliers")
+ model.score(sparkContext, sqlContext, data)
+ }
/**
@@ -78,8 +95,7 @@ object FlowSuspiciousConnectsAnalysis {
* @param inputFlowRecords raw flow records
* @return
*/
- def cleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
-
+ def filterAndSelectCleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
val cleanFlowRecordsFilter = inputFlowRecords(Hour).between(0, 23) &&
inputFlowRecords(Minute).between(0, 59) &&
inputFlowRecords(Second).between(0, 59) &&
@@ -90,9 +106,9 @@ object FlowSuspiciousConnectsAnalysis {
inputFlowRecords(DestinationPort).isNotNull &&
inputFlowRecords(Ibyt).isNotNull &&
inputFlowRecords(Ipkt).isNotNull
-
inputFlowRecords
.filter(cleanFlowRecordsFilter)
+ .select(InSchema: _*)
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index 2813f43..d13abc3 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -59,7 +59,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
@@ -120,7 +120,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- val flows : DataFrame = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(data)
logger.info("Fitting probabilistic model to data")
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/c0f80c56/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
index 035b962..82c3813 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -1,12 +1,10 @@
package org.apache.spot.netflow.model
-import org.apache.log4j.{Level, LogManager}
-import org.apache.spot.netflow.{FlowRecord, FlowSuspiciousConnectsAnalysis}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.{DataFrame, Row, SQLContext}
+import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.netflow.FlowSchema._
-import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
+import org.apache.spot.netflow.FlowSuspiciousConnectsAnalysis
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
@@ -28,7 +26,7 @@ class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with M
"filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
- val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
+ val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.filterAndSelectCleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
cleanedFlowRecords.count should be(7)
cleanedFlowRecords.schema.size should be(17)