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/06 20:00:45 UTC
[3/8] incubator-spot git commit: flow_cleaning
flow_cleaning
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/3ba71e9d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/3ba71e9d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/3ba71e9d
Branch: refs/heads/master
Commit: 3ba71e9d8181f4866d3c08513dff6462f1d77837
Parents: f5f5785
Author: nlsegerl <na...@intel.com>
Authored: Mon May 22 09:58:50 2017 -0700
Committer: nlsegerl <na...@intel.com>
Committed: Mon May 22 09:58:50 2017 -0700
----------------------------------------------------------------------
.../org/apache/spot/natetest/FlowTune.scala | 90 +++++++++++++++
.../apache/spot/netflow/FlowColumnIndex.scala | 39 -------
.../FlowSuspiciousConnectsAnalysis.scala | 2 +-
.../model/FlowSuspiciousConnectsModel.scala | 53 ++++++---
spot-ml/src/scripts/TestingFlow.scala | 43 +++++++-
.../org/apache/spot/netflow/FlowRecord.scala | 22 ++++
.../FlowSuspiciousConnectsAnalysisTest.scala | 29 +----
.../model/FlowSuspiciousConnectsModelTest.scala | 109 +++++++++++++++++++
8 files changed, 298 insertions(+), 89 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala b/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
new file mode 100644
index 0000000..49471cc
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/natetest/FlowTune.scala
@@ -0,0 +1,90 @@
+package org.apache.spot.natetest
+
+
+import org.apache.spot.netflow._
+import org.apache.spot.SuspiciousConnectsArgumentParser._
+import org.apache.log4j.{Level, LogManager, Logger}
+import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
+import org.apache.spark.sql.types.{LongType, StructField, StructType}
+import org.apache.spark.sql.{Row, SQLContext}
+import org.apache.spark.{SparkConf, SparkContext}
+/**
+ * Top level entrypoint to execute suspicious connections analysis on network data.
+ * Supported analyses:
+ * flow : netflow data
+ * dns : DNS server logs
+ * proxy : proxy server logs
+ */
+
+object FlowTune {
+
+ /**
+ * Execute suspicious connections analysis on network data.
+ *
+ * @param args Command line arguments.
+ */
+ def main(args: Array[String]) {
+
+ val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
+
+ Logger.getLogger("org").setLevel(Level.WARN)
+ Logger.getLogger("akka").setLevel(Level.OFF)
+
+ val sparkConfig = new SparkConf().setAppName("Netflow tuning")
+ val sparkContext = new SparkContext(sparkConfig)
+ val sqlContext = new SQLContext(sparkContext)
+
+ val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
+ val FeedbackFile = ""
+ val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11.baseline.scored.pqt"
+
+
+ val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
+
+ val config = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = InputParquetFile,
+ feedbackFile = FeedbackFile,
+ duplicationFactor = 1000,
+ topicCount = 20,
+ userDomain = "intel",
+ hdfsScoredConnect = ScoredParquetFile,
+ threshold = 1.0d,
+ maxResults = -1,
+ outputDelimiter = "\t",
+ ldaPRGSeed = Some(0xdeadbeef),
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+
+ val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sparkContext, sqlContext, logger, inputDataFrame).cache()
+
+ println("let me show you ten: ")
+ scoredDF.show(10)
+
+ // let's do the ROC first
+
+ val E = new BinaryClassificationMetrics(scoredDF.select("score", "is_attack").rdd.map({case (r: Row) => (r.getAs[Double](0), r.getAs[Double](1))}))
+
+ val auc = E.areaUnderROC()
+
+ println("Here's your area under the ROC curve: " + auc)
+
+ System.exit(0)
+
+ println("and now we have a dataframe with all of the ranked attacks")
+
+ val newSchema = StructType(scoredDF.schema.fields ++ Array(StructField("rank", LongType, false)))
+
+ val newRowsRDD = scoredDF.rdd.zipWithIndex().map({case (r,i) => Row.fromSeq(r.toSeq ++ Array[Any](i))})
+
+ val indexedAll = sqlContext.createDataFrame(newRowsRDD, newSchema)
+
+ val indexedAttacks = indexedAll.filter(indexedAll("is_attack") === 1)
+
+ indexedAttacks.show(20)
+ // i guess I will need to save this somewhere
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
deleted file mode 100644
index 173ab2d..0000000
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowColumnIndex.scala
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.
- */
-
-package org.apache.spot.netflow
-
-object FlowColumnIndex extends Enumeration {
- val HOUR = 4
- val MINUTE = 5
- val SECOND = 6
- val SOURCEIP = 8
- val DESTIP = 9
- val SOURCEPORT = 10
- val DESTPORT = 11
- val IPKT = 16
- val IBYT = 17
- val NUMTIME = 27
- val IBYTBIN = 28
- val IPKTYBIN = 29
- val TIMEBIN = 30
- val PORTWORD = 31
- val IPPAIR = 32
- val SOURCEWORD = 33
- val DESTWORD = 34
- }
-
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/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 4e97be1..3360f2b 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
@@ -46,7 +46,7 @@ object FlowSuspiciousConnectsAnalysis {
logger.info("Fitting probabilistic model to data")
val model =
- FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*), config.topicCount)
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, config, flows.select(InSchema: _*))
logger.info("Identifying outliers")
val scoredFlowRecords = model.score(sparkContext, sqlContext, flows)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index 95dca42..28b0457 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -155,12 +155,30 @@ object FlowSuspiciousConnectsModel {
val ModelColumns = ModelSchema.fieldNames.toList.map(col)
- def trainNewModel(sparkContext: SparkContext,
- sqlContext: SQLContext,
- logger: Logger,
- config: SuspiciousConnectsConfig,
- inputRecords: DataFrame,
- topicCount: Int): FlowSuspiciousConnectsModel = {
+
+ def cleanData(flows: DataFrame): DataFrame = {
+
+ val legalFlowsFilter = flows(Hour).between(0, 23) &&
+ flows(Minute).between(0, 59) &&
+ flows(Second).between(0, 59) &&
+ flows(TimeReceived).isNotNull &&
+ flows(SourceIP).isNotNull &&
+ flows(DestinationIP).isNotNull &&
+ flows(SourcePort).isNotNull &&
+ flows(DestinationPort).isNotNull &&
+ flows(Ibyt).isNotNull &&
+ flows(Ipkt).isNotNull
+
+ flows.filter(legalFlowsFilter)
+ }
+
+
+ def trainModel(sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger,
+ config: SuspiciousConnectsConfig,
+ inputRecords: DataFrame): FlowSuspiciousConnectsModel = {
+
logger.info("Training netflow suspicious connects model from " + config.inputPath)
@@ -281,17 +299,18 @@ object FlowSuspiciousConnectsModel {
.map({ case ((ip, word), count) => SpotLDAInput(ip, word, count) })
- val SpotLDAOutput(ipToTopicMix, wordToPerTopicProb) = SpotLDAWrapper.runLDA(sparkContext,
- sqlContext,
- ipWordCounts,
- config.topicCount,
- logger,
- config.ldaPRGSeed,
- config.ldaAlpha,
- config.ldaBeta,
- config.ldaMaxiterations)
-
- new FlowSuspiciousConnectsModel(topicCount,
+ val SpotLDAOutput(ipToTopicMix, wordToPerTopicProb) =
+ SpotLDAWrapper.runLDA(sparkContext,
+ sqlContext,
+ ipWordCounts,
+ config.topicCount,
+ logger,
+ config.ldaPRGSeed,
+ config.ldaAlpha,
+ config.ldaBeta,
+ config.ldaMaxiterations)
+
+ new FlowSuspiciousConnectsModel(config.topicCount,
ipToTopicMix,
wordToPerTopicProb,
config.useProtocol,
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/scripts/TestingFlow.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/scripts/TestingFlow.scala b/spot-ml/src/scripts/TestingFlow.scala
index 9876945..a713fba 100644
--- a/spot-ml/src/scripts/TestingFlow.scala
+++ b/spot-ml/src/scripts/TestingFlow.scala
@@ -1,13 +1,22 @@
import org.apache.spot.netflow._
+import org.apache.spot.netflow.FlowSchema._
import org.apache.spot.SuspiciousConnectsArgumentParser._
import org.apache.log4j.{Level, LogManager, Logger}
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
+
val logger = LogManager.getLogger("SuspiciousConnects: Flow Model Testing")
-val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
+Logger.getLogger("org").setLevel(Level.WARN)
+Logger.getLogger("akka").setLevel(Level.OFF)
+
+val InputParquetFile = "/user/duxbury/flow/hive/y=2016/m=05/d=05"
+//val InputParquetFile = "/user/nlsegerl/2016.11.11.synthetic.batch1.labeled.pqt"
val FeedbackFile = ""
-val ScoredParquetFile = "/user/nlsegerl/test/2016.11.11/baseline.scored.pqt"
+val ScoredParquetFile = "/user/nlsegerl/test/2016.05.05.baseline.scored.pqt"
val inputDataFrame = sqlContext.read.parquet(InputParquetFile)
@@ -16,19 +25,41 @@ val config = SuspiciousConnectsConfig(analysis = "flow",
inputPath = InputParquetFile,
feedbackFile = FeedbackFile,
duplicationFactor = 1000,
- topicCount = 20,
+ topicCount = 10,
userDomain = "intel",
hdfsScoredConnect = ScoredParquetFile,
threshold = 1.0d,
maxResults = -1,
outputDelimiter = "\t",
ldaPRGSeed = Some(0xdeadbeef),
- ldaMaxiterations = 20,
+ ldaMaxiterations = 5,
ldaAlpha = 1.02,
ldaBeta = 1.001)
-val scoredDF = FlowSuspiciousConnectsAnalysis.run(config, sc, sqlContext, logger, inputDataFrame).cache()
-// there are two big tasks .... rank the attacks and calculate the area under the ROC curve
+logger.info("Starting flow suspicious connects analysis.")
+
+val cleanFlows : DataFrame = FlowSuspiciousConnectsModel.cleanData(inputDataFrame) // ? should we cache here?
+
+logger.info("Fitting model to data")
+
+val model = FlowSuspiciousConnectsModel.trainModel(sc, sqlContext, logger, config, cleanFlows)
+
+logger.info("Assigning scores to flow records")
+
+val scoredFlowRecords = model.score(sc, sqlContext, cleanFlows)
+
+logger.info("Selecting most suspicious flow records")
+
+val orderedFlowRecords = scoredFlowRecords.filter(scoredFlowRecords(Score) < 0.0001).orderBy(Score)
+
+
+
+orderedFlowRecords.show(10)
+
+
+val df = orderedFlowRecords
+
+
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
new file mode 100644
index 0000000..8349c9e
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowRecord.scala
@@ -0,0 +1,22 @@
+package org.apache.spot.netflow
+
+/**
+ * Created by nlsegerl on 5/16/17.
+ */
+case class FlowRecord(treceived: String,
+ tryear: Int,
+ trmonth: Int,
+ trday: Int,
+ trhour: Int,
+ trminute: Int,
+ trsec: Int,
+ tdur: Float,
+ sip: String,
+ dip: String,
+ sport: Int,
+ dport: Int,
+ proto: String,
+ ipkt: Int,
+ ibyt: Int,
+ opkt: Int,
+ obyt: Int)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/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 a92d052..c16fb13 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
@@ -26,23 +26,7 @@ import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
-case class FlowRecord(treceived: String,
- tryear: Int,
- trmonth: Int,
- trday: Int,
- trhour: Int,
- trminute: Int,
- trsec: Int,
- tdur: Float,
- sip: String,
- dip: String,
- sport: Int,
- dport: Int,
- proto: String,
- ipkt: Int,
- ibyt: Int,
- opkt: Int,
- obyt: Int)
+
class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Matchers {
@@ -75,13 +59,12 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
- import FlowSuspiciousConnectsAnalysis.{cleanFlowRecords, InSchema}
- val flows : DataFrame = cleanFlowRecords(data)
+ val flows : DataFrame = FlowSuspiciousConnectsModel.cleanData(data)
logger.info("Fitting probabilistic model to data")
val model =
- FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, testConfig, flows.select(InSchema: _*), testConfig.topicCount)
+ FlowSuspiciousConnectsModel.trainModel(sparkContext, sqlContext, logger, testConfig, flows)
logger.info("Identifying outliers")
val scoredData = model.score(sparkContext, sqlContext, flows)
@@ -104,13 +87,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
- "filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
-
- val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis.cleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
- cleanedFlowRecords.count should be(5)
- cleanedFlowRecords.schema.size should be(17)
- }
"filterAndSelectInvalidFlowRecords" should "return invalid records" in {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/3ba71e9d/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
new file mode 100644
index 0000000..0f52713
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModelTest.scala
@@ -0,0 +1,109 @@
+package org.apache.spot.netflow.model
+
+import org.apache.log4j.{Level, LogManager}
+import org.apache.spot.netflow.FlowRecord
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{DataFrame, 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.testutils.TestingSparkContextFlatSpec
+import org.scalatest.Matchers
+
+class FlowSuspiciousConnectsModelTest extends TestingSparkContextFlatSpec with Matchers {
+
+ val testConfig = SuspiciousConnectsConfig(analysis = "flow",
+ inputPath = "",
+ feedbackFile = "",
+ duplicationFactor = 1,
+ topicCount = 20,
+ hdfsScoredConnect = "",
+ threshold = 1.0d,
+ maxResults = 1000,
+ outputDelimiter = "\t",
+ ldaPRGSeed = None,
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
+
+ "filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
+
+ val cleanedFlowRecords = FlowSuspiciousConnectsModel.cleanData(testFlowRecords.inputFlowRecordsDF)
+
+ cleanedFlowRecords.count should be(5)
+ cleanedFlowRecords.schema.size should be(17)
+ }
+
+ def testFlowRecords = new {
+ val sqlContext = new SQLContext(sparkContext)
+
+ val inputFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 24, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 60, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 60, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq(null, 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, null, "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", null, 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", null, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, null, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", null, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, null, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, null, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, null),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0))
+ .map(row => Row.fromSeq(row))))
+
+ val inputFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField))
+
+ val inputFlowRecordsDF = sqlContext.createDataFrame(inputFlowRecordsRDD, inputFlowRecordsSchema)
+
+ val scoredFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, -1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0000005),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.05),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0001))
+ .map(row => Row.fromSeq(row))))
+
+ val scoredFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField,
+ ScoreField))
+
+ val scoredFlowRecordsDF = sqlContext.createDataFrame(scoredFlowRecordsRDD, scoredFlowRecordsSchema)
+ }
+}