diff --git a/spot-ml/SPARKCONF.md b/spot-ml/SPARKCONF.md index 3f57ce36..e7fd8f9e 100644 --- a/spot-ml/SPARKCONF.md +++ b/spot-ml/SPARKCONF.md @@ -1,14 +1,14 @@ -##Spark Configuration +## Spark Configuration -oni-ml main component uses Spark and Spark SQL to analyze network events and produce a list of least probable events -or most suspicious. +spot-ml main component uses Spark and Spark SQL to analyze network events and those considered the most unlikely +or most suspicious. -To run oni-ml with its best performance and scalability, it will probably be necessary to configure Yarn, Spark and Spot. Here are our recommended settings. +To run spot-ml with its best performance and scalability, it will probably be necessary to configure Yarn, Spark and Spot. Here are our recommended settings. ### General Yarn tuning -oni-ml Spark application has been developed and tested on CDH [Yarn](http://spark.apache.org/docs/latest/running-on-yarn.html) -clusters. Careful tuning of the Yarn cluster may be necessary before analyzing large amounts of data with oni-ml. +spot-ml Spark application has been developed and tested on CDH [Yarn](http://spark.apache.org/docs/latest/running-on-yarn.html) +clusters. Careful tuning of the Yarn cluster may be necessary before analyzing large amounts of data with spot-ml. For small data sets, under 100 GB parquet files, default _Yarn_ configurations should be enough but if users try to analyze hundreds of gigabytes of data in parquet format it's probable that it don't work; _Yarn_ most likely will start killing @@ -27,10 +27,10 @@ one try before getting any results. When running _Spark_ on _Yarn_ users can set up a set of properties in order to get the best performance and consume resources in a more effective way. Since not all clusters are the same and not all users are planning to have the same capacity of computation, we have created -variables that users need to configure before running oni-ml. +variables that users need to configure before running spot-ml. -After installing oni-setup users will find the [spot.conf](https://github.com/Open-Network-Insight/open-network-insight/blob/spot/spot-setup/spot.conf) -file under /etc folder. This file contains all the required configuration to run oni-ml, as explained in [INSTALL.md](https://github.com/Open-Network-Insight/oni-ml/blob/master/INSTALL.md). In this +After installing spot-setup users will find the [spot.conf](https://github.com/apache/incubator-spot/blob/master/spot-setup/spot.conf) +file under /etc folder. This file contains all the required configuration to run spot-ml, as explained in [INSTALL.md](https://github.com/apache/incubator-spot/blob/master/spot-ml/INSTALL.md). In this file exist a section for _Spark_ properties, below is the explanation for each of those variables: SPK_EXEC='' ---> Maximumn number of executors @@ -39,16 +39,18 @@ file exist a section for _Spark_ properties, below is the explanation for each o SPK_DRIVER_MAX_RESULTS='' ---> Maximumn driver results in MB or GB i.e. 8g SPK_EXEC_CORES='' ---> Cores per executor i.e. 4 SPK_DRIVER_MEM_OVERHEAD='' ---> Driver memory overhead in MB i.e. 3047. Note that there is no "m" at the end. - SPAK_EXEC_MEM_OVERHEAD='' ---> Executor memory overhead in MB i.e. 3047. Note that there is no "m" at the end. + SPK_EXEC_MEM_OVERHEAD='' ---> Executor memory overhead in MB i.e. 3047. Note that there is no "m" at the end. + SPK_AUTO_BRDCST_JOIN_THR='10485760' ---> Spark's spark.sql.autoBroadcastJoinThreshold. Default is 10MB, increase this value to make Spark broadcast tables larger than 10 MB and speed up joins. + PRECISION='64' ---> Indicates whether spot-ml is to use 64 bit floating point numbers or 32 bit floating point numbers when representing certain probability distributions. -Besides the variables in duxbay.conf, users can modify the rest of the properties in ml_ops.sh based on their needs. +Besides the variables in spot.conf, users can modify the rest of the properties in ml_ops.sh based on their needs. #### Setting Spark properties - After _Yarn_ cluster has been tuned the next step is to set Spark properties assigning the right values to duxbay.conf _Spark_ + After _Yarn_ cluster has been tuned the next step is to set Spark properties assigning the right values to spot.conf _Spark_ variables. -#####Number of Executors, Executor Memory, Executor Cores and Executor Memory Overhead +##### Number of Executors, Executor Memory, Executor Cores and Executor Memory Overhead The first thing users need to know is how to set the number of executors and the memory per executor as well as the number of cores. To get that number, users should know the available total memory per node after _Yarn_ tuning, this total memory is determined by _yarn.nodemanager.resource.memory-mb_ @@ -87,16 +89,16 @@ Is recommended to keep a close relation between cores and executor memory. Lastly, for overhead memory we recommend to use something between 8% and 10% of executor memory. -Following the example, the values for the _Spark_ variables in duxbay.conf would look like this: +Following the example, the values for the _Spark_ variables in spot.conf would look like this: SPK_EXEC='43' SPK_EXEC_MEM='30475m' SPK_EXEC_CORES='6' SPAK_EXEC_MEM_OVERHEAD='3047' -#####Driver Memory, Driver Maximum Results and Driver Memory Overhead +##### Driver Memory, Driver Maximum Results and Driver Memory Overhead - oni-ml application executes actions such as _.collect_, _.orderBy_, _.saveAsTextFile_ so we recommend to assign a + spot-ml application executes actions such as _.collect_, _.orderBy_, _.saveAsTextFile_ so we recommend to assign a considerable amount of memory for the driver. The same way, driver maximum results should be enough for the serialized results. @@ -120,34 +122,64 @@ and 50 GB. Driver maximum results should be something equal or bigger than 8 GB. SPK_DRIVER_MEM='30475m' SPK_DRIVER_MAX_RESULTS='8g' SPK_DRIVER_MEM_OVERHEAD='3047' - - ![DriverMemory](https://raw.githubusercontent.com/Open-Network-Insight/oni-docs/master/images/Driver%20Memory.png) + _Representation of memory allocation in driver node._ - For more information about Spark properties click [here](http://spark.apache.org/docs/latest/configuration.html). -###Known Spark error messages running oni-ml + ##### Spark autoBroadcastJoinThreshold in spot-ml + + After Spark LDA runs, Topics Matrix and Topics Distribution are joined with the original data set i.e. NetFlow records, DNS records or Proxy records to determine the probability of each event to happen. This joining process is similar to join a big data set and a lookup table. In this case, the big data set is the entire set of records, and the lookup table is a dictionary of documents and probabilities per topic or words and probabilities per topic. + + Because of the possible diversity of documents/IPs, the lookup table containing document probability distribution can grow to something bigger than 10 MB. Taking in account that 10 MB is Spark's default auto broadcast threshold for joins, a join with a lookup table bigger than that threshold will result in the execution of a traditional join with lots of shuffling. + +The correct setting of SPK_AUTO_BRDCST_JOIN_THR and PRECISION can help to always broadcast document probability distribution lookup table and avoid slow joins. + +As a first step, users need to decide whether they want to change from 64 bit floating point probabilities to 32 bit floating point probabilities; if users decide to change from 64 to 32 bit, the document probability distribution lookup table will be half the size and more easily broadcasted. + +If users want to cut payload memory consumption roughly in half, they should set the precision option to 32. + + >PRECISION='32' + + If users prefer to keep 64 bit floating point numbers, they should set precision option to 64 (default). + + >PRECISION='64' + +Given the approximate number of distinct IPs in every batch or data set being analyzed, users should set SPK_AUTO_BRDCST_JOIN_THR to something that can fit the document probability distribution lookup table. + +For instance, if a user knows there can be 2,000,000 distinct IP addresses and is using 20 Topics, the document probability distribution lookup table can grow to something like 190 bytes per row if using PRECISION as 64 bit and 110 bytes per row if using 32 bit option. + +> Document probability distribution lookup table record example: + +>(192.169.111.110, [0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05, 0.05] + +In that case, users should set auto broadcast join threshold to something that can fit 365 MB (380000000 Bytes) for 64 bit floating precision numbers or 210 MB (220000000 Bytes) for 32 bit floating precision numbers. + +> PRECISION='32' + +> SPK_AUTO_BRDCST_JOIN_THR='220000000' + +### Known Spark error messages running spot-ml -####Out Of Memory Error +#### Out Of Memory Error This issue includes _java.lang.OutOfMemoryError: Java heap space_ and _java.lang.OutOfMemoryError : GC overhead limit exceeded_. When users get OOME can be for many different issues but we have identified a couple of reasons for this -error in oni-ml. +error in spot-ml. -The main reason for this error in oni-ml can be when the ML algorithm returns large results for _word probabilities per topic_. +The main reason for this error in spot-ml can be when the ML algorithm returns large results for _word probabilities per topic_. Since ML algorithm results are broadcast, each executor needs more memory. Another possible reason for this error is driver is running out of memory, try increasing driver memory. -####Container killed by Yarn for exceeding memory limits. _X.Y_ GB of _X_ GB physical memory used. Consider boosting spark.yarn.executor memoryOverhead +#### Container killed by Yarn for exceeding memory limits. _X.Y_ GB of _X_ GB physical memory used. Consider boosting spark.yarn.executor memoryOverhead This issue is caused by certain operations, mainly during the join of _document probabilities per topic_ with the rest of the data - scoring stage. If users receive this error they should try with increasing memory overhead up to 10% of executor memory or increase executors memory. -####org.apache.spark.serializer.KryoSerializer +#### org.apache.spark.serializer.KryoSerializer KryoSerializer can cause issues if property _spark.kryoserializer.buffer.max_ is not enough for the data being serialized. Try increasing memory up to 2 GB but keeping in mind the total of the available memory. diff --git a/spot-ml/ml_ops.sh b/spot-ml/ml_ops.sh index 7bf98d75..339899d2 100755 --- a/spot-ml/ml_ops.sh +++ b/spot-ml/ml_ops.sh @@ -36,7 +36,6 @@ if [[ "${#FDATE}" != "8" || -z "${DSOURCE}" ]]; then exit fi - # read in variables (except for date) from etc/.conf file # note: FDATE and DSOURCE *must* be defined prior sourcing this conf file @@ -90,6 +89,7 @@ time spark-submit --class "org.apache.spot.SuspiciousConnects" \ --conf spark.dynamicAllocation.enabled=true \ --conf spark.executor.cores=${SPK_EXEC_CORES} \ --conf spark.executor.memory=${SPK_EXEC_MEM} \ + --conf spark.sql.autoBroadcastJoinThreshold=${SPK_AUTO_BRDCST_JOIN_THR} \ --conf "spark.executor.extraJavaOptions=-XX:MaxPermSize=512M -XX:PermSize=512M" \ --conf spark.kryoserializer.buffer.max=512m \ --conf spark.yarn.am.waitTime=100s \ @@ -104,5 +104,6 @@ time spark-submit --class "org.apache.spot.SuspiciousConnects" \ --threshold ${TOL} \ --maxresults ${MAXRESULTS} \ --ldamaxiterations 20 \ + --precision ${PRECISION} \ $USER_DOMAIN_CMD diff --git a/spot-ml/ml_test.sh b/spot-ml/ml_test.sh index b8b2642e..8fed8375 100755 --- a/spot-ml/ml_test.sh +++ b/spot-ml/ml_test.sh @@ -35,6 +35,14 @@ LPATH=${LUSER}/ml/${DSOURCE}/test HPATH=${HUSER}/${DSOURCE}/test/scored_results # prepare parameters pipeline stages +# pass the user domain designation if not empty + +if [ ! -z $USER_DOMAIN ] ; then + USER_DOMAIN_CMD="--userdomain $USER_DOMAIN" +else + USER_DOMAIN_CMD='' +fi + FEEDBACK_PATH=${LPATH}/${DSOURCE}_scores.csv HDFS_SCORED_CONNECTS=${HPATH}/scores @@ -47,20 +55,17 @@ hdfs dfs -rm -R -f ${HDFS_SCORED_CONNECTS} time spark-submit --class "org.apache.spot.SuspiciousConnects" \ --master yarn-client \ --driver-memory ${SPK_DRIVER_MEM} \ + --num-executors ${SPK_EXEC} \ --conf spark.driver.maxResultSize=${SPK_DRIVER_MAX_RESULTS} \ --conf spark.driver.maxPermSize=512m \ - --conf spark.driver.cores=1 \ --conf spark.dynamicAllocation.enabled=true \ - --conf spark.dynamicAllocation.minExecutors=1 \ - --conf spark.dynamicAllocation.maxExecutors=${SPK_EXEC} \ --conf spark.executor.cores=${SPK_EXEC_CORES} \ --conf spark.executor.memory=${SPK_EXEC_MEM} \ + --conf spark.sql.autoBroadcastJoinThreshold=${SPK_AUTO_BRDCST_JOIN_THR} \ --conf "spark.executor.extraJavaOptions=-XX:MaxPermSize=512M -XX:PermSize=512M" \ - --conf spark.shuffle.io.preferDirectBufs=false \ --conf spark.kryoserializer.buffer.max=512m \ - --conf spark.shuffle.service.enabled=true \ - --conf spark.yarn.am.waitTime=1000000 \ - --conf spark.yarn.driver.memoryOverhead=${SPK_DRIVER_MEM_OVERHEAD} \ + --conf spark.yarn.am.waitTime=100s \ + --conf spark.yarn.am.memoryOverhead=${SPK_DRIVER_MEM_OVERHEAD} \ --conf spark.yarn.executor.memoryOverhead=${SPK_EXEC_MEM_OVERHEAD} target/scala-2.10/spot-ml-assembly-1.1.jar \ --analysis ${DSOURCE} \ --input ${RAWDATA_PATH} \ @@ -70,4 +75,6 @@ time spark-submit --class "org.apache.spot.SuspiciousConnects" \ --scored ${HDFS_SCORED_CONNECTS} \ --threshold ${TOL} \ --maxresults ${MAXRESULTS} \ - --ldamaxiterations 11 \ No newline at end of file + --ldamaxiterations 20 \ + --precision ${PRECISION} \ + $USER_DOMAIN_CMD \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala index 1966dec1..f5472e67 100644 --- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala +++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsArgumentParser.scala @@ -17,27 +17,14 @@ package org.apache.spot +import org.apache.spot.utilities.{FloatPointPrecisionUtility, FloatPointPrecisionUtility32, FloatPointPrecisionUtility64} + /** * Parses arguments for the suspicious connections analysis. */ object SuspiciousConnectsArgumentParser { - case class SuspiciousConnectsConfig(analysis: String = "", - inputPath: String = "", - feedbackFile: String = "", - duplicationFactor: Int = 1, - topicCount: Int = 20, - userDomain: String = "", - hdfsScoredConnect: String = "", - threshold: Double = 1.0d, - maxResults: Int = -1, - outputDelimiter: String = "\t", - ldaPRGSeed: Option[Long] = None, - ldaMaxiterations: Int = 20, - ldaAlpha: Double = 1.02, - ldaBeta: Double = 1.001) - val parser: scopt.OptionParser[SuspiciousConnectsConfig] = new scopt.OptionParser[SuspiciousConnectsConfig]("LDA") { head("LDA Process", "1.1") @@ -99,5 +86,28 @@ object SuspiciousConnectsArgumentParser { opt[Double]("ldabeta").optional().valueName("float64"). action((x, c) => c.copy(ldaBeta = x)). text("topic concentration for lda, default 1.001") + + opt[Int]("precision").optional().valueName("int"). + action((x, c) => c.copy(precisionUtility = x match { + case 32 => FloatPointPrecisionUtility32 + case 64 => FloatPointPrecisionUtility64 + case _ => FloatPointPrecisionUtility64 + })) } + + case class SuspiciousConnectsConfig(analysis: String = "", + inputPath: String = "", + feedbackFile: String = "", + duplicationFactor: Int = 1, + topicCount: Int = 20, + userDomain: String = "", + hdfsScoredConnect: String = "", + threshold: Double = 1.0d, + maxResults: Int = -1, + outputDelimiter: String = "\t", + ldaPRGSeed: Option[Long] = None, + ldaMaxiterations: Int = 20, + ldaAlpha: Double = 1.02, + ldaBeta: Double = 1.001, + precisionUtility: FloatPointPrecisionUtility = FloatPointPrecisionUtility64) } diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala index 93e938ed..97cde68b 100644 --- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala +++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala @@ -18,6 +18,7 @@ package org.apache.spot import org.apache.spark.broadcast.Broadcast +import org.apache.spot.utilities.FloatPointPrecisionUtility import org.apache.spot.utilities.data.validation.InvalidDataHandler /** @@ -25,26 +26,27 @@ import org.apache.spot.utilities.data.validation.InvalidDataHandler * Assumes that distribution of words is independent of the IP when conditioned on the topic * and performs a simple sum over a partition of the space by topic. * - * @param topicCount Number of topics produced by the topic modelling analysis. - * @param ipToTopicMixBC Broadcast of map assigning IPs to topic mixes. + * @param topicCount Number of topics produced by the topic modelling analysis. * @param wordToPerTopicProbBC Broadcast of map assigning words to per-topic conditional probability. */ class SuspiciousConnectsScoreFunction(topicCount: Int, - ipToTopicMixBC: Broadcast[Map[String, Array[Double]]], wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable { - def score(ip: String, word: String): Double = { + def score[P <: FloatPointPrecisionUtility](precisionUtility: P) + (documentTopicMix: Seq[precisionUtility.TargetType], word: String): Double = { - val zeroProb = Array.fill(topicCount) { 0d } + val zeroProb = Array.fill(topicCount) { + 0d + } - if(word == InvalidDataHandler.WordError){ + if (word == InvalidDataHandler.WordError) { InvalidDataHandler.ScoreError } else { // If either the ip or the word key value cannot be found it means that it was not seen in training. - val topicGivenDocProbs = ipToTopicMixBC.value.getOrElse(ip, zeroProb) - val wordGivenTopicProbs = wordToPerTopicProbBC.value.getOrElse(word, zeroProb) + val wordGivenTopicProbabilities = wordToPerTopicProbBC.value.getOrElse(word, zeroProb) + val documentTopicMixDouble: Seq[Double] = precisionUtility.toDoubles(documentTopicMix) - topicGivenDocProbs.zip(wordGivenTopicProbs) + documentTopicMixDouble.zip(wordGivenTopicProbabilities) .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc }) .sum } diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala index 94e38513..17e0b424 100644 --- a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala +++ b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala @@ -37,6 +37,22 @@ import org.apache.spot.utilities.data.validation.{InvalidDataHandler => dataVali object DNSSuspiciousConnectsAnalysis { + val DefaultQueryClass = "unknown" + val DefaultQueryType = -1 + val DefaultQueryResponseCode = -1 + val InStructType = StructType(List(TimestampField, UnixTimestampField, FrameLengthField, ClientIPField, + QueryNameField, QueryClassField, QueryTypeField, QueryResponseCodeField)) + val InSchema = InStructType.fieldNames.map(col) + val OutSchema = StructType( + List(TimestampField, + UnixTimestampField, + FrameLengthField, + ClientIPField, + QueryNameField, + QueryClassField, + QueryTypeField, + QueryResponseCodeField, + ScoreField)).fieldNames.map(col) /** * Run suspicious connections analysis on DNS log data. @@ -77,7 +93,6 @@ object DNSSuspiciousConnectsAnalysis { dataValidation.showAndSaveCorruptRecords(corruptDNSRecords, config.hdfsScoredConnect, logger) } - /** * Identify anomalous DNS log entries in in the provided data frame. * @@ -99,10 +114,9 @@ object DNSSuspiciousConnectsAnalysis { DNSSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, data, config.topicCount) logger.info("Identifying outliers") - model.score(sparkContext, sqlContext, data, config.userDomain) + model.score(sparkContext, sqlContext, data, config.userDomain, config.precisionUtility) } - /** * * @param inputDNSRecords raw DNS records. @@ -136,7 +150,6 @@ object DNSSuspiciousConnectsAnalysis { .na.fill(DefaultQueryResponseCode, Seq(QueryResponseCode)) } - /** * * @param inputDNSRecords raw DNS records. @@ -167,7 +180,6 @@ object DNSSuspiciousConnectsAnalysis { .select(InSchema: _*) } - /** * * @param scoredDNSRecords scored DNS records. @@ -183,6 +195,8 @@ object DNSSuspiciousConnectsAnalysis { scoredDNSRecords.filter(filteredDNSRecordsFilter) } + assert(ModelSchema.fields.forall(InStructType.fields.contains(_))) + /** * * @param scoredDNSRecords scored DNS records. @@ -198,27 +212,4 @@ object DNSSuspiciousConnectsAnalysis { } - - val DefaultQueryClass = "unknown" - val DefaultQueryType = -1 - val DefaultQueryResponseCode = -1 - - val InStructType = StructType(List(TimestampField, UnixTimestampField, FrameLengthField, ClientIPField, - QueryNameField, QueryClassField, QueryTypeField, QueryResponseCodeField)) - - val InSchema = InStructType.fieldNames.map(col) - - assert(ModelSchema.fields.forall(InStructType.fields.contains(_))) - - val OutSchema = StructType( - List(TimestampField, - UnixTimestampField, - FrameLengthField, - ClientIPField, - QueryNameField, - QueryClassField, - QueryTypeField, - QueryResponseCodeField, - ScoreField)).fieldNames.map(col) - } \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSScoreFunction.scala index 77c1c439..f98f7cb2 100644 --- a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSScoreFunction.scala +++ b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSScoreFunction.scala @@ -20,21 +20,21 @@ package org.apache.spot.dns.model import org.apache.spark.broadcast.Broadcast import org.apache.spot.SuspiciousConnectsScoreFunction import org.apache.spot.dns.DNSWordCreation +import org.apache.spot.utilities.FloatPointPrecisionUtility /** * Estimate the probabilities of network events using a [[DNSSuspiciousConnectsModel]] * - * @param frameLengthCuts Delimeters used to define binning for frame length field - * @param timeCuts Delimeters used to define binning for time field - * @param subdomainLengthCuts Delimeters used to define binning for subdomain length field - * @param entropyCuts Delimeters used to define binning for entropy field - * @param numberPeriodsCuts Delimeters used to define binning for number of periods of subdomain field - * @param topicCount Number of topics used for the LDA model - * @param ipToTopicMixBC Topic mixes learned by the LDA model for each IP in the data + * @param frameLengthCuts Delimeters used to define binning for frame length field + * @param timeCuts Delimeters used to define binning for time field + * @param subdomainLengthCuts Delimeters used to define binning for subdomain length field + * @param entropyCuts Delimeters used to define binning for entropy field + * @param numberPeriodsCuts Delimeters used to define binning for number of periods of subdomain field + * @param topicCount Number of topics used for the LDA model * @param wordToPerTopicProbBC Word mixes for each of the topics learned by the LDA model - * @param topDomainsBC Alexa top one million list of domains. - * @param userDomain Domain associated to network data (example: 'intel') + * @param topDomainsBC Alexa top one million list of domains. + * @param userDomain Domain associated to network data (example: 'intel') */ class DNSScoreFunction(frameLengthCuts: Array[Double], timeCuts: Array[Double], @@ -42,31 +42,31 @@ class DNSScoreFunction(frameLengthCuts: Array[Double], entropyCuts: Array[Double], numberPeriodsCuts: Array[Double], topicCount: Int, - ipToTopicMixBC: Broadcast[Map[String, Array[Double]]], wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]], topDomainsBC: Broadcast[Set[String]], userDomain: String) extends Serializable { val suspiciousConnectsScoreFunction = - new SuspiciousConnectsScoreFunction(topicCount, ipToTopicMixBC, wordToPerTopicProbBC) + new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) val dnsWordCreator = new DNSWordCreation(frameLengthCuts, - timeCuts, - subdomainLengthCuts, - entropyCuts, - numberPeriodsCuts, - topDomainsBC, - userDomain) + timeCuts, + subdomainLengthCuts, + entropyCuts, + numberPeriodsCuts, + topDomainsBC, + userDomain) - def score(timeStamp: String, - unixTimeStamp: Long, - frameLength: Int, - clientIP: String, - queryName: String, - queryClass: String, - queryType: Int, - queryResponseCode: Int): Double = { + def score[P <: FloatPointPrecisionUtility](precisionUtility: P)(timeStamp: String, + unixTimeStamp: Long, + frameLength: Int, + clientIP: String, + queryName: String, + queryClass: String, + queryType: Int, + queryResponseCode: Int, + documentTopicMix: Seq[precisionUtility.TargetType]): Double = { val word = dnsWordCreator.dnsWord(timeStamp, unixTimeStamp, @@ -77,6 +77,6 @@ class DNSScoreFunction(frameLengthCuts: Array[Double], queryType, queryResponseCode) - suspiciousConnectsScoreFunction.score(clientIP, word) + suspiciousConnectsScoreFunction.score(precisionUtility)(documentTopicMix, word) } } \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala index df88eb1d..088f4fd6 100644 --- a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala +++ b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala @@ -28,10 +28,11 @@ import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig import org.apache.spot.dns.DNSSchema._ import org.apache.spot.dns.DNSWordCreation import org.apache.spot.lda.SpotLDAWrapper -import org.apache.spot.lda.SpotLDAWrapper.{SpotLDAInput, SpotLDAOutput} +import org.apache.spot.lda.SpotLDAWrapper._ +import org.apache.spot.lda.SpotLDAWrapperSchema._ import org.apache.spot.utilities.DomainProcessor.DomainInfo +import org.apache.spot.utilities._ import org.apache.spot.utilities.data.validation.InvalidDataHandler -import org.apache.spot.utilities.{CountryCodes, DomainProcessor, Quantiles, TopDomains} import scala.util.{Failure, Success, Try} @@ -60,7 +61,7 @@ import scala.util.{Failure, Success, Try} * @param inEntropyCuts Quantile cut-offs for discretizing the subdomain entropy in word construction. */ class DNSSuspiciousConnectsModel(inTopicCount: Int, - inIpToTopicMix: Map[String, Array[Double]], + inIpToTopicMix: DataFrame, inWordToPerTopicProb: Map[String, Array[Double]], inTimeCuts: Array[Double], inFrameLengthCuts: Array[Double], @@ -88,13 +89,12 @@ class DNSSuspiciousConnectsModel(inTopicCount: Int, * @return Dataframe with a column named [[org.apache.spot.dns.DNSSchema.Score]] that contains the * probability estimated for the network event at that row */ - def score(sc: SparkContext, sqlContext: SQLContext, inDF: DataFrame, userDomain: String): DataFrame = { + def score(sc: SparkContext, sqlContext: SQLContext, inDF: DataFrame, userDomain: String + , precisionUtility: FloatPointPrecisionUtility): DataFrame = { val topDomainsBC = sc.broadcast(TopDomains.TopDomains) - val ipToTopicMixBC = sc.broadcast(ipToTopicMix) val wordToPerTopicProbBC = sc.broadcast(wordToPerTopicProb) - val scoreFunction = new DNSScoreFunction(frameLengthCuts, timeCuts, @@ -102,7 +102,6 @@ class DNSSuspiciousConnectsModel(inTopicCount: Int, entropyCuts, numberPeriodsCuts, topicCount, - ipToTopicMixBC, wordToPerTopicProbBC, topDomainsBC, userDomain) @@ -115,17 +114,24 @@ class DNSSuspiciousConnectsModel(inTopicCount: Int, queryName: String, queryClass: String, queryType: Int, - queryResponseCode: Int) => - scoreFunction.score(timeStamp, + queryResponseCode: Int, + documentTopicMix: Seq[precisionUtility.TargetType]) => + scoreFunction.score(precisionUtility)(timeStamp, unixTimeStamp, frameLength, clientIP, queryName, queryClass, queryType, - queryResponseCode)) - - inDF.withColumn(Score, scoringUDF(DNSSuspiciousConnectsModel.modelColumns: _*)) + queryResponseCode, + documentTopicMix)) + + inDF + .join(org.apache.spark.sql.functions.broadcast(ipToTopicMix), inDF(ClientIP) === ipToTopicMix(DocumentName), + "left_outer") + .selectExpr(inDF.schema.fieldNames :+ TopicProbabilityMix: _*) + .withColumn(Score, scoringUDF(DNSSuspiciousConnectsModel.modelColumns :+ col(TopicProbabilityMix): _*)) + .drop(TopicProbabilityMix) } } @@ -275,7 +281,7 @@ object DNSSuspiciousConnectsModel { .map({ case ((ipDst, word), count) => SpotLDAInput(ipDst, word, count) }) - val SpotLDAOutput(ipToTopicMixDF, wordToPerTopicProb) = SpotLDAWrapper.runLDA(sparkContext, + val SpotLDAOutput(ipToTopicMix, wordToPerTopicProb) = SpotLDAWrapper.runLDA(sparkContext, sqlContext, ipDstWordCounts, config.topicCount, @@ -283,19 +289,8 @@ object DNSSuspiciousConnectsModel { config.ldaPRGSeed, config.ldaAlpha, config.ldaBeta, - config.ldaMaxiterations) - - // Since DNS is still broadcasting ip to topic mix, we need to convert data frame to Map[String, Array[Double]] - val ipToTopicMix = ipToTopicMixDF - .rdd - .map({ case (ipToTopicMixRow: Row) => ipToTopicMixRow.toSeq.toArray }) - .map({ - case (ipToTopicMixSeq) => (ipToTopicMixSeq(0).asInstanceOf[String], ipToTopicMixSeq(1).asInstanceOf[Seq[Double]] - .toArray) - }) - .collectAsMap - .toMap - + config.ldaMaxiterations, + config.precisionUtility) new DNSSuspiciousConnectsModel(topicCount, ipToTopicMix, @@ -335,9 +330,6 @@ object DNSSuspiciousConnectsModel { sqlContext.createDataFrame(domainStatsRDD, DomainStatsSchema) } - - case class TempFields(topDomainClass: Int, subdomainLength: Integer, subdomainEntropy: Double, numPeriods: Integer) - /** * * @param countryCodesBC Broadcast of the country codes set. @@ -360,4 +352,6 @@ object DNSSuspiciousConnectsModel { subdomainEntropy = subdomainEntropy, numPeriods = numPeriods) } + + case class TempFields(topDomainClass: Int, subdomainLength: Integer, subdomainEntropy: Double, numPeriods: Integer) } \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/lda/SpotLDAWrapper.scala b/spot-ml/src/main/scala/org/apache/spot/lda/SpotLDAWrapper.scala index d58cecb9..9e882e02 100644 --- a/spot-ml/src/main/scala/org/apache/spot/lda/SpotLDAWrapper.scala +++ b/spot-ml/src/main/scala/org/apache/spot/lda/SpotLDAWrapper.scala @@ -24,8 +24,8 @@ import org.apache.spark.mllib.linalg.{Matrix, Vector, Vectors} import org.apache.spark.rdd.RDD import org.apache.spark.sql.functions._ import org.apache.spark.sql.{DataFrame, Row, SQLContext} - -import SpotLDAWrapperSchema._ +import org.apache.spot.lda.SpotLDAWrapperSchema._ +import org.apache.spot.utilities.FloatPointPrecisionUtility import scala.collection.immutable.Map @@ -40,11 +40,6 @@ import scala.collection.immutable.Map object SpotLDAWrapper { - case class SpotLDAInput(doc: String, word: String, count: Int) extends Serializable - - case class SpotLDAOutput(docToTopicMix: DataFrame, wordResults: Map[String, Array[Double]]) - - def runLDA(sparkContext: SparkContext, sqlContext: SQLContext, docWordCount: RDD[SpotLDAInput], @@ -53,7 +48,8 @@ object SpotLDAWrapper { ldaSeed: Option[Long], ldaAlpha: Double, ldaBeta: Double, - maxIterations: Int): SpotLDAOutput = { + maxIterations: Int, + precisionUtility: FloatPointPrecisionUtility): SpotLDAOutput = { import sqlContext.implicits._ @@ -123,7 +119,7 @@ object SpotLDAWrapper { //Create doc results from vector: convert docID back to string, convert vector of probabilities to array val docToTopicMixDF = - formatSparkLDADocTopicOutput(docTopicDist, documentDictionary, sqlContext) + formatSparkLDADocTopicOutput(docTopicDist, documentDictionary, sqlContext, precisionUtility) documentDictionary.unpersist() @@ -186,20 +182,27 @@ object SpotLDAWrapper { wordProbs.zipWithIndex.map({ case (topicProbs, wordInd) => (wordMap(wordInd), topicProbs) }).toMap } - def formatSparkLDADocTopicOutput(docTopDist: RDD[(Long, Vector)], documentDictionary: DataFrame, sqlContext: SQLContext): + def formatSparkLDADocTopicOutput(docTopDist: RDD[(Long, Vector)], documentDictionary: DataFrame, sqlContext: + SQLContext, precisionUtility: FloatPointPrecisionUtility): DataFrame = { import sqlContext.implicits._ val topicDistributionToArray = udf((topicDistribution: Vector) => topicDistribution.toArray) val documentToTopicDistributionDF = docTopDist.toDF(DocumentNumber, TopicProbabilityMix) - documentToTopicDistributionDF + val documentToTopicDistributionArray = documentToTopicDistributionDF .join(documentDictionary, documentToTopicDistributionDF(DocumentNumber) === documentDictionary(DocumentNumber)) .drop(documentDictionary(DocumentNumber)) .drop(documentToTopicDistributionDF(DocumentNumber)) .select(DocumentName, TopicProbabilityMix) .withColumn(TopicProbabilityMixArray, topicDistributionToArray(documentToTopicDistributionDF(TopicProbabilityMix))) .selectExpr(s"$DocumentName AS $DocumentName", s"$TopicProbabilityMixArray AS $TopicProbabilityMix") + + precisionUtility.castColumn(documentToTopicDistributionArray, TopicProbabilityMix) } + case class SpotLDAInput(doc: String, word: String, count: Int) extends Serializable + + case class SpotLDAOutput(docToTopicMix: DataFrame, wordResults: Map[String, Array[Double]]) + } \ No newline at end of file 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 b9421aea..71212e47 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 @@ -36,6 +36,43 @@ import org.apache.spot.utilities.data.validation.{InvalidDataHandler => dataVali object FlowSuspiciousConnectsAnalysis { + val InSchema = StructType(List(TimeReceivedField, + YearField, + MonthField, + DayField, + HourField, + MinuteField, + SecondField, + DurationField, + SourceIPField, + DestinationIPField, + SourcePortField, + DestinationPortField, + ProtocolField, + IpktField, + IbytField, + OpktField, + ObytField)).fieldNames.map(col) + val OutSchema = StructType( + List(TimeReceivedField, + YearField, + MonthField, + DayField, + HourField, + MinuteField, + SecondField, + DurationField, + SourceIPField, + DestinationIPField, + SourcePortField, + DestinationPortField, + ProtocolField, + IpktField, + IbytField, + OpktField, + ObytField, + ScoreField)).fieldNames.map(col) + def run(config: SuspiciousConnectsConfig, sparkContext: SparkContext, sqlContext: SQLContext, logger: Logger, inputFlowRecords: DataFrame) = { @@ -85,7 +122,7 @@ object FlowSuspiciousConnectsAnalysis { FlowSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, data, config.topicCount) logger.info("Identifying outliers") - model.score(sparkContext, sqlContext, data) + model.score(sparkContext, sqlContext, data, config.precisionUtility) } /** @@ -164,43 +201,4 @@ object FlowSuspiciousConnectsAnalysis { } - - val InSchema = StructType(List(TimeReceivedField, - YearField, - MonthField, - DayField, - HourField, - MinuteField, - SecondField, - DurationField, - SourceIPField, - DestinationIPField, - SourcePortField, - DestinationPortField, - ProtocolField, - IpktField, - IbytField, - OpktField, - ObytField)).fieldNames.map(col) - - val OutSchema = StructType( - List(TimeReceivedField, - YearField, - MonthField, - DayField, - HourField, - MinuteField, - SecondField, - DurationField, - SourceIPField, - DestinationIPField, - SourcePortField, - DestinationPortField, - ProtocolField, - IpktField, - IbytField, - OpktField, - ObytField, - ScoreField)).fieldNames.map(col) - } \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala index 2ee29a8a..61f42e6c 100644 --- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala +++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala @@ -18,8 +18,9 @@ package org.apache.spot.netflow.model import org.apache.spark.broadcast.Broadcast +import org.apache.spot.SuspiciousConnectsScoreFunction import org.apache.spot.netflow.{FlowWordCreator, FlowWords} -import org.apache.spot.utilities.data.validation.InvalidDataHandler +import org.apache.spot.utilities.FloatPointPrecisionUtility /** @@ -43,6 +44,9 @@ class FlowScoreFunction(timeCuts: Array[Double], val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts) + val suspiciousConnectsScoreFunction = + new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) + /** * Estimate the probability of a netflow connection as distributed from the source IP and from the destination IP * and assign it the least of these two values. @@ -60,47 +64,25 @@ class FlowScoreFunction(timeCuts: Array[Double], * @param dstTopicMix topic mix assigned of destination IP * @return Minium of probability of this word from the source IP and probability of this word from the dest IP. */ - def score(hour: Int, - minute: Int, - second: Int, - srcIP: String, - dstIP: String, - srcPort: Int, - dstPort: Int, - ipkt: Long, - ibyt: Long, - srcTopicMix: Seq[Double], - dstTopicMix: Seq[Double]): Double = { + def score[P <: FloatPointPrecisionUtility](precisionUtility: P)(hour: Int, + minute: Int, + second: Int, + srcIP: String, + dstIP: String, + srcPort: Int, + dstPort: Int, + ipkt: Long, + ibyt: Long, + srcTopicMix: Seq[precisionUtility.TargetType], + dstTopicMix: Seq[precisionUtility.TargetType]): Double = { val FlowWords(srcWord, dstWord) = flowWordCreator.flowWords(hour: Int, minute: Int, second: Int, srcPort: Int, dstPort: Int, ipkt: Long, ibyt: Long) - val zeroProb = Array.fill(topicCount) { - 0.0 - } - - /** WordError indicates there was a problem creating a word and should not be used for scoring. - * * - * A null value for srcTopicMix or dstTopicMix indicated the ip (source or dest respectively) - * were not seen in training. - */ - if (srcWord == InvalidDataHandler.WordError || dstWord == InvalidDataHandler.WordError) { - InvalidDataHandler.ScoreError - } else if (srcTopicMix == null || dstTopicMix == null) { - 0.0 - } else { - - val scoreOfConnectionFromSrcIP = srcTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(srcWord, zeroProb)) - .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc }) - .sum - - val scoreOfConnectionsFromDstIP = dstTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(dstWord, zeroProb)) - .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc }) - .sum - - Math.min(scoreOfConnectionFromSrcIP, scoreOfConnectionsFromDstIP) + val srcIPScore = suspiciousConnectsScoreFunction.score(precisionUtility)(srcTopicMix, srcWord) + val dstIPScore = suspiciousConnectsScoreFunction.score(precisionUtility)(dstTopicMix, dstWord) - } + Math.min(srcIPScore, dstIPScore) } } 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 b5fdda53..3efd886f 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 @@ -29,8 +29,8 @@ import org.apache.spot.lda.SpotLDAWrapper.{SpotLDAInput, SpotLDAOutput} import org.apache.spot.lda.SpotLDAWrapperSchema._ import org.apache.spot.netflow.FlowSchema._ import org.apache.spot.netflow.FlowWordCreator -import org.apache.spot.utilities.Quantiles import org.apache.spot.utilities.data.validation.InvalidDataHandler +import org.apache.spot.utilities.{FloatPointPrecisionUtility, Quantiles} import scala.util.{Failure, Success, Try} @@ -65,7 +65,8 @@ class FlowSuspiciousConnectsModel(topicCount: Int, ibytCuts: Array[Double], ipktCuts: Array[Double]) { - def score(sc: SparkContext, sqlContext: SQLContext, flowRecords: DataFrame): DataFrame = { + def score(sc: SparkContext, sqlContext: SQLContext, flowRecords: DataFrame, + precisionUtility: FloatPointPrecisionUtility): DataFrame = { val wordToPerTopicProbBC = sc.broadcast(wordToPerTopicProb) @@ -75,13 +76,13 @@ class FlowSuspiciousConnectsModel(topicCount: Int, */ val dataWithSrcTopicMix = { - val recordsWithSrcIPTopicMixes = flowRecords.join(ipToTopicMix, + val recordsWithSrcIPTopicMixes = flowRecords.join(org.apache.spark.sql.functions.broadcast(ipToTopicMix), flowRecords(SourceIP) === ipToTopicMix(DocumentName), "left_outer") val schemaWithSrcTopicMix = flowRecords.schema.fieldNames :+ TopicProbabilityMix val dataWithSrcIpProb: DataFrame = recordsWithSrcIPTopicMixes.selectExpr(schemaWithSrcTopicMix: _*) .withColumnRenamed(TopicProbabilityMix, SrcIpTopicMix) - val recordsWithIPTopicMixes = dataWithSrcIpProb.join(ipToTopicMix, + val recordsWithIPTopicMixes = dataWithSrcIpProb.join(org.apache.spark.sql.functions.broadcast(ipToTopicMix), dataWithSrcIpProb(DestinationIP) === ipToTopicMix(DocumentName), "left_outer") val schema = dataWithSrcIpProb.schema.fieldNames :+ TopicProbabilityMix recordsWithIPTopicMixes.selectExpr(schema: _*).withColumnRenamed(TopicProbabilityMix, DstIpTopicMix) @@ -103,9 +104,9 @@ class FlowSuspiciousConnectsModel(topicCount: Int, dstPort: Int, ipkt: Long, ibyt: Long, - srcIpTopicMix: Seq[Double], - dstIpTopicMix: Seq[Double]) => - scoreFunction.score(hour, + srcIpTopicMix: Seq[precisionUtility.TargetType], + dstIpTopicMix: Seq[precisionUtility.TargetType]) => + scoreFunction.score(precisionUtility)(hour, minute, second, srcIP, @@ -253,7 +254,8 @@ object FlowSuspiciousConnectsModel { config.ldaPRGSeed, config.ldaAlpha, config.ldaBeta, - config.ldaMaxiterations) + config.ldaMaxiterations, + config.precisionUtility) new FlowSuspiciousConnectsModel(topicCount, ipToTopicMix, diff --git a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysis.scala index cbc3eb05..133dc916 100644 --- a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysis.scala +++ b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysis.scala @@ -31,6 +31,51 @@ import org.apache.spot.utilities.data.validation.{InvalidDataHandler => dataVali */ object ProxySuspiciousConnectsAnalysis { + val DefaultUserAgent = "-" + val DefaultResponseContentType = "-" + val InSchema = StructType( + List(DateField, + TimeField, + ClientIPField, + HostField, + ReqMethodField, + UserAgentField, + ResponseContentTypeField, + DurationField, + UserNameField, + WebCatField, + RefererField, + RespCodeField, + URIPortField, + URIPathField, + URIQueryField, + ServerIPField, + SCBytesField, + CSBytesField, + FullURIField)).fieldNames.map(col) + val OutSchema = StructType( + List(DateField, + TimeField, + ClientIPField, + HostField, + ReqMethodField, + UserAgentField, + ResponseContentTypeField, + DurationField, + UserNameField, + WebCatField, + RefererField, + RespCodeField, + URIPortField, + URIPathField, + URIQueryField, + ServerIPField, + SCBytesField, + CSBytesField, + FullURIField, + WordField, + ScoreField)).fieldNames.map(col) + /** * Run suspicious connections analysis on proxy data. * @@ -91,7 +136,7 @@ object ProxySuspiciousConnectsAnalysis { val model = ProxySuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, data) logger.info("Identifying outliers") - model.score(sparkContext, data) + model.score(sparkContext, data, config.precisionUtility) } /** @@ -159,51 +204,4 @@ object ProxySuspiciousConnectsAnalysis { .filter(corruptProxyRecordsFilter) .select(OutSchema: _*) } - - val DefaultUserAgent = "-" - val DefaultResponseContentType = "-" - - val InSchema = StructType( - List(DateField, - TimeField, - ClientIPField, - HostField, - ReqMethodField, - UserAgentField, - ResponseContentTypeField, - DurationField, - UserNameField, - WebCatField, - RefererField, - RespCodeField, - URIPortField, - URIPathField, - URIQueryField, - ServerIPField, - SCBytesField, - CSBytesField, - FullURIField)).fieldNames.map(col) - - val OutSchema = StructType( - List(DateField, - TimeField, - ClientIPField, - HostField, - ReqMethodField, - UserAgentField, - ResponseContentTypeField, - DurationField, - UserNameField, - WebCatField, - RefererField, - RespCodeField, - URIPortField, - URIPathField, - URIQueryField, - ServerIPField, - SCBytesField, - CSBytesField, - FullURIField, - WordField, - ScoreField)).fieldNames.map(col) } \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala index 74dae769..ef28d2a4 100644 --- a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala +++ b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala @@ -27,6 +27,7 @@ import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig import org.apache.spot.SuspiciousConnectsScoreFunction import org.apache.spot.lda.SpotLDAWrapper import org.apache.spot.lda.SpotLDAWrapper.{SpotLDAInput, SpotLDAOutput} +import org.apache.spot.lda.SpotLDAWrapperSchema._ import org.apache.spot.proxy.ProxySchema._ import org.apache.spot.utilities._ import org.apache.spot.utilities.data.validation.InvalidDataHandler @@ -39,7 +40,7 @@ import org.apache.spot.utilities.data.validation.InvalidDataHandler * @param wordToPerTopicProb Maps each word to a vector measuring Prob[word | topic] for each topic. */ class ProxySuspiciousConnectsModel(topicCount: Int, - ipToTopicMIx: Map[String, Array[Double]], + ipToTopicMIx: DataFrame, wordToPerTopicProb: Map[String, Array[Double]]) { /** @@ -50,7 +51,7 @@ class ProxySuspiciousConnectsModel(topicCount: Int, * (as defined in ProxySchema object). * @return Dataframe with Score column added. */ - def score(sc: SparkContext, dataFrame: DataFrame): DataFrame = { + def score(sc: SparkContext, dataFrame: DataFrame, precisionUtility: FloatPointPrecisionUtility): DataFrame = { val topDomains: Broadcast[Set[String]] = sc.broadcast(TopDomains.TopDomains) @@ -71,15 +72,18 @@ class ProxySuspiciousConnectsModel(topicCount: Int, dataFrame(UserAgent), dataFrame(RespCode))) - val ipToTopicMixBC = sc.broadcast(ipToTopicMIx) val wordToPerTopicProbBC = sc.broadcast(wordToPerTopicProb) + val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) - val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, ipToTopicMixBC, wordToPerTopicProbBC) + def udfScoreFunction = udf((documentTopicMix: Seq[precisionUtility.TargetType], word: String) => + scoreFunction.score(precisionUtility)(documentTopicMix, word)) - - def udfScoreFunction = udf((ip: String, word: String) => scoreFunction.score(ip, word)) - wordedDataFrame.withColumn(Score, udfScoreFunction(wordedDataFrame(ClientIP), wordedDataFrame(Word))) + wordedDataFrame + .join(org.apache.spark.sql.functions.broadcast(ipToTopicMIx), dataFrame(ClientIP) === ipToTopicMIx(DocumentName), "left_outer") + .selectExpr(wordedDataFrame.schema.fieldNames :+ TopicProbabilityMix: _*) + .withColumn(Score, udfScoreFunction(col(TopicProbabilityMix), col(Word))) + .drop(TopicProbabilityMix) } } @@ -141,7 +145,7 @@ object ProxySuspiciousConnectsModel { agentToCount) - val SpotLDAOutput(ipToTopicMixDF, wordResults) = SpotLDAWrapper.runLDA(sparkContext, + val SpotLDAOutput(ipToTopicMix, wordResults) = SpotLDAWrapper.runLDA(sparkContext, sqlContext, docWordCount, config.topicCount, @@ -149,20 +153,8 @@ object ProxySuspiciousConnectsModel { config.ldaPRGSeed, config.ldaAlpha, config.ldaBeta, - config.ldaMaxiterations) - - - // Since Proxy is still broadcasting ip to topic mix, we need to convert data frame to Map[String, Array[Double]] - val ipToTopicMix = ipToTopicMixDF - .rdd - .map({ case (ipToTopicMixRow: Row) => ipToTopicMixRow.toSeq.toArray }) - .map({ - case (ipToTopicMixSeq) => (ipToTopicMixSeq(0).asInstanceOf[String], ipToTopicMixSeq(1).asInstanceOf[Seq[Double]] - .toArray) - }) - .collectAsMap - .toMap - + config.ldaMaxiterations, + config.precisionUtility) new ProxySuspiciousConnectsModel(config.topicCount, ipToTopicMix, wordResults) diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/CountryCodes.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/CountryCodes.scala index 4b5e9915..e54ca6c3 100644 --- a/spot-ml/src/main/scala/org/apache/spot/utilities/CountryCodes.scala +++ b/spot-ml/src/main/scala/org/apache/spot/utilities/CountryCodes.scala @@ -17,7 +17,6 @@ package org.apache.spot.utilities - object CountryCodes { val CountryCodes = Set("ac", "ad", "ae", "af", "ag", "ai", "al", "am", "an", "ao", "aq", "ar", "as", "at", "au", diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/DomainProcessor.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/DomainProcessor.scala index 4f5c9714..d0fabf81 100644 --- a/spot-ml/src/main/scala/org/apache/spot/utilities/DomainProcessor.scala +++ b/spot-ml/src/main/scala/org/apache/spot/utilities/DomainProcessor.scala @@ -42,25 +42,6 @@ object DomainProcessor extends Serializable { val TopLevelDomainNames = Set("com", "org", "net", "int", "edu", "gov", "mil") val None = "None" - - /** - * Commonly extracted domain features. - * @param domain Domain (if any) of a url. - * @param topDomain Numerical class of domain: 2 for Intel, 1 for Alexa top domains, 0 for others. - * @param subdomain Subdomain (if any) in the url. - * @param subdomainLength Length of the subdomain. 0 if there is none. - * @param subdomainEntropy Entropy of the subdomain viewed as a distribution on its character set. - * 0 if there is no subdomain. - * @param numPeriods Number of periods + 1 in the url. (Number of sub-strings where url is split by periods.) - */ - case class DomainInfo(domain: String, - topDomain: Int, - subdomain: String, - subdomainLength: Int, - subdomainEntropy: Double, - numPeriods: Int) - - /** * Extract domain info from a url. * @param url Incoming url. @@ -95,7 +76,6 @@ object DomainProcessor extends Serializable { DomainInfo(domain, topDomainClass, subdomain, subdomainLength, subdomainEntropy, numParts) } - /** * * @param url Url from which to extract domain. @@ -163,4 +143,22 @@ object DomainProcessor extends Serializable { urlComponents } } + + /** + * Commonly extracted domain features. + * + * @param domain Domain (if any) of a url. + * @param topDomain Numerical class of domain: 2 for Intel, 1 for Alexa top domains, 0 for others. + * @param subdomain Subdomain (if any) in the url. + * @param subdomainLength Length of the subdomain. 0 if there is none. + * @param subdomainEntropy Entropy of the subdomain viewed as a distribution on its character set. + * 0 if there is no subdomain. + * @param numPeriods Number of periods + 1 in the url. (Number of sub-strings where url is split by periods.) + */ + case class DomainInfo(domain: String, + topDomain: Int, + subdomain: String, + subdomainLength: Int, + subdomainEntropy: Double, + numPeriods: Int) } diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/FloatPointPrecisionUtility.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/FloatPointPrecisionUtility.scala new file mode 100644 index 00000000..a4da7a0f --- /dev/null +++ b/spot-ml/src/main/scala/org/apache/spot/utilities/FloatPointPrecisionUtility.scala @@ -0,0 +1,108 @@ +/* + * 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.utilities + +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions._ + +/** + * PrecisionUtility will transform a number from Double to Float if precision option is set to 32 bit, + * if default or 64 bit is selected, it will just return the same number type Double. + * + */ +sealed trait FloatPointPrecisionUtility extends Serializable { + + type TargetType + + /** + * Converts a number into the precision type; it can be Float (32) or Double (64). + * + * @param double a number to convert from Double to Target type. + * @return + */ + def toTargetType(double: Double): TargetType + + /** + * Converts back an Iterable of numbers from TargetType to Double. <% determines that it can process any object that + * extends from Iterable i.e. List, Vector, Seq, Queue, etc. + * + * @param targetTypeIterable an iterable that needs to be converted each element from TargetType to Double + * @return + */ + def toDoubles[A <% Traversable[TargetType], B <% Traversable[Double]](targetTypeIterable: A): B + + /** + * Converts a DataFrame column from Seq[Double] to a Seq[TargetType]. + * + * @param dataFrame a DataFrame containing a column to be converted from Double to the TargetType + * @param columnName the name of the column to convert, the column should be Seq[Double] + * @return + */ + def castColumn(dataFrame: DataFrame, columnName: String): DataFrame + +} + +/** + * PrecisionUtility implementation for Float. + * Will convert numbers from Double to Float and back to Double. + * + */ +object FloatPointPrecisionUtility32 extends FloatPointPrecisionUtility { + + type TargetType = Float + + def toTargetType(double: Double): Float = double.toFloat + + def toDoubles[A <% Traversable[Float], B <% Traversable[Double]](targetTypeIterable: A): B = + targetTypeIterable.map(_.toDouble).asInstanceOf[B] + + + private val convertUDF = udf((doubles: Seq[Double]) => { + doubles.map(double => toTargetType(double)) + }) + + def castColumn(dataFrame: DataFrame, columnName: String): DataFrame = { + val TempColumn = "temp_column" + + dataFrame.withColumn(TempColumn, convertUDF(dataFrame(columnName))) + .drop(columnName) + .withColumnRenamed(TempColumn, columnName) + } + +} + +/** + * PrecisionUtility implementation for Double. + * This implementation will receive and send the same value, it won't do any transformation. + * + */ +object FloatPointPrecisionUtility64 extends FloatPointPrecisionUtility { + + type TargetType = Double + + // For this implementation it will just return the same value without any transformation. + def toTargetType(double: Double): Double = double + + // Since Double is the default data type, this code is going to return the same array without any transformation. + def toDoubles[A <% Traversable[Double], B <% Traversable[Double]](targetTypeIterable: A): B = + targetTypeIterable.asInstanceOf[B] + + // Since Double is the default data type, this code won't actually do any calculation but instead just return the + // same DataFrame. + def castColumn(dataFrame: DataFrame, columnName: String): DataFrame = dataFrame +} \ No newline at end of file diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/Quantiles.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/Quantiles.scala index 07bad9c3..5494ab5d 100644 --- a/spot-ml/src/main/scala/org/apache/spot/utilities/Quantiles.scala +++ b/spot-ml/src/main/scala/org/apache/spot/utilities/Quantiles.scala @@ -27,41 +27,25 @@ import scala.math._ object Quantiles extends Serializable { + val DECILES = Array(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0) + val QUINTILES = Array(0.2, 0.4, 0.6, 0.8, 1.0) + /** - * Compute the empirical cumulative distribution function, - * cf. [[https://en.wikipedia.org/wiki/Empirical_distribution_function]] + * Compute the deciles of a distribution. * - * @param data An RDD of doubles. - * @return RDD[(Double,Double)] where each pair is of the form (value, ecdf at value) - * That is, each pair is a value and the fraction of the input data less-than-or-equal to the value. + * @param data RDD[Double] Incoming data. + * @return Array[Double]. The deciles of the distribution. */ - def computeEcdf(data: RDD[Double]): RDD[(Double, Double)] = { - - // pair each distinct value with the number of times it appears in the data, then sort by the data values - val valueCountPairs = data.map(v => (v, 1L)).reduceByKey(_ + _).sortByKey().persist() - - // for each partition, sum the counts of all values in the partition - val countsPerPartition: Array[Double] = valueCountPairs.mapPartitionsWithIndex { - case (_, partition) => Iterator(partition.map({ case (_, count) => count }).sum.toDouble) - }.collect() - - val totalCount = countsPerPartition.sum - - // pair each value v with the sum of counts of all values <= v in the data - val valueCountLEQPairs : RDD[(Double, Double)] = valueCountPairs.mapPartitionsWithIndex { - case (index, partition) => - val countInPrecedingPartitions = countsPerPartition.take(index).sum - val p = partition.scanLeft((0.0, countInPrecedingPartitions))({ case ((_, countOfLEQValues), (value, countOfValue)) => - (value, countOfLEQValues + countOfValue)}) + def computeDeciles(data: RDD[Double]): Array[Double] = computeQuantiles(data, DECILES) - // first element is an extraneous zero and must be dropped - p.drop(1) - } - valueCountPairs.unpersist() - // normalize counts by total number entries in the data to obtain the ecdf - valueCountLEQPairs.map({case (value, countToLeftOfValue) => (value, countToLeftOfValue / totalCount)}) - } + /** + * Compute the quintiles of a distribution. + * + * @param data RDD[Double] Incoming data. + * @return Array[Double]. The quintiles of the distribution. + */ + def computeQuintiles(data: RDD[Double]): Array[Double] = computeQuantiles(data, QUINTILES) /** * Compute the quantiles for a given dataset and array of thresholds for the cumulative distribution. @@ -98,25 +82,41 @@ object Quantiles extends Serializable { computeEcdf(data).aggregate(initialCutoffs)(addDataPointToKnownCutoffs, mergeCutoffs) } - /** - * Compute the deciles of a distribution. + * Compute the empirical cumulative distribution function, + * cf. [[https://en.wikipedia.org/wiki/Empirical_distribution_function]] * - * @param data RDD[Double] Incoming data. - * @return Array[Double]. The deciles of the distribution. + * @param data An RDD of doubles. + * @return RDD[(Double,Double)] where each pair is of the form (value, ecdf at value) + * That is, each pair is a value and the fraction of the input data less-than-or-equal to the value. */ - def computeDeciles(data: RDD[Double]): Array[Double] = computeQuantiles(data, DECILES) - val DECILES = Array(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0) - /** - * Compute the quintiles of a distribution. - * - * @param data RDD[Double] Incoming data. - * @return Array[Double]. The quintiles of the distribution. - */ - def computeQuintiles(data: RDD[Double]): Array[Double] = computeQuantiles(data, QUINTILES) - val QUINTILES = Array(0.2, 0.4, 0.6, 0.8, 1.0) + def computeEcdf(data: RDD[Double]): RDD[(Double, Double)] = { + + // pair each distinct value with the number of times it appears in the data, then sort by the data values + val valueCountPairs = data.map(v => (v, 1L)).reduceByKey(_ + _).sortByKey().persist() + + // for each partition, sum the counts of all values in the partition + val countsPerPartition: Array[Double] = valueCountPairs.mapPartitionsWithIndex { + case (_, partition) => Iterator(partition.map({ case (_, count) => count }).sum.toDouble) + }.collect() + + val totalCount = countsPerPartition.sum + + // pair each value v with the sum of counts of all values <= v in the data + val valueCountLEQPairs : RDD[(Double, Double)] = valueCountPairs.mapPartitionsWithIndex { + case (index, partition) => + val countInPrecedingPartitions = countsPerPartition.take(index).sum + val p = partition.scanLeft((0.0, countInPrecedingPartitions))({ case ((_, countOfLEQValues), (value, countOfValue)) => + (value, countOfLEQValues + countOfValue)}) + // first element is an extraneous zero and must be dropped + p.drop(1) + } + valueCountPairs.unpersist() + // normalize counts by total number entries in the data to obtain the ecdf + valueCountLEQPairs.map({case (value, countToLeftOfValue) => (value, countToLeftOfValue / totalCount)}) + } def bin(value: Double, cuts: Array[Double]) : Int = { cuts.indexWhere(cut => value <= cut) diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/TimeUtilities.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/TimeUtilities.scala index 003664cb..830e93e9 100644 --- a/spot-ml/src/main/scala/org/apache/spot/utilities/TimeUtilities.scala +++ b/spot-ml/src/main/scala/org/apache/spot/utilities/TimeUtilities.scala @@ -17,7 +17,6 @@ package org.apache.spot.utilities - object TimeUtilities { diff --git a/spot-ml/src/test/scala/org/apache/spot/SpotLDAWrapperTest.scala b/spot-ml/src/test/scala/org/apache/spot/SpotLDAWrapperTest.scala deleted file mode 100644 index 8baaf348..00000000 --- a/spot-ml/src/test/scala/org/apache/spot/SpotLDAWrapperTest.scala +++ /dev/null @@ -1,145 +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 - -import org.apache.log4j.{Level, LogManager} -import org.apache.spark.mllib.linalg.{Matrices, Vector, Vectors} -import org.apache.spark.rdd.RDD -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.{DataFrame, Row, SQLContext} -import org.apache.spot.lda.SpotLDAWrapperSchema._ -import org.apache.spot.lda.SpotLDAWrapper -import org.apache.spot.lda.SpotLDAWrapper._ -import org.apache.spot.testutils.TestingSparkContextFlatSpec -import org.scalatest.Matchers - -import scala.collection.immutable.Map - -class SpotLDAWrapperTest extends TestingSparkContextFlatSpec with Matchers { - - val ldaAlpha = 1.02 - val ldaBeta = 1.001 - val ldaMaxiterations = 20 - - "SparkLDA" should "handle an extremely unbalanced two word doc" in { - val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") - logger.setLevel(Level.WARN) - - val catFancy = SpotLDAInput("pets", "cat", 1) - val dogWorld = SpotLDAInput("pets", "dog", 999) - - val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) - val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, ldaMaxiterations) - - val topicMixDF = out.docToTopicMix - - var topicMix = - topicMixDF.filter(topicMixDF(DocumentName) === "pets").select(TopicProbabilityMix).first().toSeq(0).asInstanceOf[Seq[Double]].toArray - val catTopics = out.wordResults("cat") - val dogTopics = out.wordResults("dog") - - Math.abs(topicMix(0) * catTopics(0) + topicMix(1) * catTopics(1)) should be < 0.01 - Math.abs(0.999 - (topicMix(0) * dogTopics(0) + topicMix(1) * dogTopics(1))) should be < 0.01 - } - - "SparkLDA" should "handle distinct docs on distinct words" in { - val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") - logger.setLevel(Level.WARN) - val catFancy = SpotLDAInput("cat fancy", "cat", 1) - val dogWorld = SpotLDAInput("dog world", "dog", 1) - - val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) - val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, ldaMaxiterations) - - val topicMixDF = out.docToTopicMix - var dogTopicMix: Array[Double] = - topicMixDF.filter(topicMixDF(DocumentName) === "dog world").select(TopicProbabilityMix).first().toSeq(0).asInstanceOf[Seq[Double]].toArray - - val catTopicMix: Array[Double] = - topicMixDF.filter(topicMixDF(DocumentName) === "cat fancy").select(TopicProbabilityMix).first().toSeq(0).asInstanceOf[Seq[Double]].toArray - - val catTopics = out.wordResults("cat") - val dogTopics = out.wordResults("dog") - - Math.abs(1 - (catTopicMix(0) * catTopics(0) + catTopicMix(1) * catTopics(1))) should be < 0.01 - Math.abs(1 - (dogTopicMix(0) * dogTopics(0) + dogTopicMix(1) * dogTopics(1))) should be < 0.01 - } - - "formatSparkLDAInput" should "return input in RDD[(Long, Vector)] (collected as Array for testing) format. The index " + - "is the docID, values are the vectors of word occurrences in that doc" in { - - - val documentWordData = sparkContext.parallelize(Seq(SpotLDAInput("192.168.1.1", "333333_7.0_0.0_1.0", 8), - SpotLDAInput("10.10.98.123", "1111111_6.0_3.0_5.0", 4), - SpotLDAInput("66.23.45.11", "-1_43_7.0_2.0_6.0", 2), - SpotLDAInput("192.168.1.1", "-1_80_6.0_1.0_1.0", 5))) - - val wordDictionary = Map("333333_7.0_0.0_1.0" -> 0, "1111111_6.0_3.0_5.0" -> 1, "-1_43_7.0_2.0_6.0" -> 2, "-1_80_6.0_1.0_1.0" -> 3) - - val documentDictionary: DataFrame = sqlContext.createDataFrame(documentWordData - .map({ case SpotLDAInput(doc, word, count) => doc }) - .distinct - .zipWithIndex.map({case (d,c) => Row(d,c)}), StructType(List(DocumentNameField, DocumentNumberField))) - - - val sparkLDAInput: RDD[(Long, Vector)] = SpotLDAWrapper.formatSparkLDAInput(documentWordData, documentDictionary, wordDictionary, sqlContext) - val sparkLDAInArr: Array[(Long, Vector)] = sparkLDAInput.collect() - - sparkLDAInArr shouldBe Array((0, Vectors.sparse(4, Array(0, 3), Array(8.0, 5.0))), (2, Vectors.sparse(4, Array(2), Array(2.0))), (1, Vectors.sparse(4, Array(1), Array(4.0)))) - } - - "formatSparkLDADocTopicOuptut" should "return RDD[(String,Array(Double))] after converting doc results from vector: " + - "convert docID back to string, convert vector of probabilities to array" in { - - - val documentWordData = sparkContext.parallelize(Seq(SpotLDAInput("192.168.1.1", "333333_7.0_0.0_1.0", 8), - SpotLDAInput("10.10.98.123", "1111111_6.0_3.0_5.0", 4), - SpotLDAInput("66.23.45.11", "-1_43_7.0_2.0_6.0", 2), - SpotLDAInput("192.168.1.1", "-1_80_6.0_1.0_1.0", 5))) - - val documentDictionary: DataFrame = sqlContext.createDataFrame(documentWordData - .map({ case SpotLDAInput(doc, word, count) => doc }) - .distinct - .zipWithIndex.map({case (d,c) => Row(d,c)}), StructType(List(DocumentNameField, DocumentNumberField))) - - val docTopicDist: RDD[(Long, Vector)] = sparkContext.parallelize(Array((0.toLong, Vectors.dense(0.15, 0.3, 0.5, 0.05)), (1.toLong, - Vectors.dense(0.25, 0.15, 0.4, 0.2)), (2.toLong, Vectors.dense(0.4, 0.1, 0.3, 0.2)))) - - val sparkDocRes: DataFrame = formatSparkLDADocTopicOutput(docTopicDist, documentDictionary, sqlContext) - - val documents = sparkDocRes.select(DocumentName).map(documentName => documentName.toString.replaceAll("\\[", "").replaceAll("\\]", "")).collect() - - documents(0) should be("10.10.98.123") - documents(1) should be("192.168.1.1") - documents(2) should be("66.23.45.11") - } - - "formatSparkLDAWordOutput" should "return Map[Int,String] after converting word matrix to sequence, wordIDs back to strings, and sequence of probabilities to array" in { - val testMat = Matrices.dense(4, 4, Array(0.5, 0.2, 0.05, 0.25, 0.25, 0.1, 0.15, 0.5, 0.1, 0.4, 0.25, 0.25, 0.7, 0.2, 0.02, 0.08)) - - val wordDictionary = Map("-1_23.0_7.0_7.0_4.0" -> 3, "23.0_7.0_7.0_4.0" -> 0, "333333.0_7.0_7.0_4.0" -> 2, "80.0_7.0_7.0_4.0" -> 1) - val revWordMap: Map[Int, String] = wordDictionary.map(_.swap) - - val sparkWordRes = formatSparkLDAWordOutput(testMat, revWordMap) - - sparkWordRes should contain key ("23.0_7.0_7.0_4.0") - sparkWordRes should contain key ("80.0_7.0_7.0_4.0") - sparkWordRes should contain key ("333333.0_7.0_7.0_4.0") - sparkWordRes should contain key ("-1_23.0_7.0_7.0_4.0") - } -} \ No newline at end of file diff --git a/spot-ml/src/test/scala/org/apache/spot/SuspiciousConnectsScoreFunctionTest.scala b/spot-ml/src/test/scala/org/apache/spot/SuspiciousConnectsScoreFunctionTest.scala new file mode 100644 index 00000000..08f782e7 --- /dev/null +++ b/spot-ml/src/test/scala/org/apache/spot/SuspiciousConnectsScoreFunctionTest.scala @@ -0,0 +1,79 @@ +/* + * 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 + +import org.apache.spark.broadcast.Broadcast +import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.apache.spot.utilities.{FloatPointPrecisionUtility32, FloatPointPrecisionUtility64} +import org.scalatest.Matchers + +/** + * Created by rabarona on 5/17/17. + */ +class SuspiciousConnectsScoreFunctionTest extends TestingSparkContextFlatSpec with Matchers { + + "score" should "return score of type Double with document probabilities of type Double" in { + + val wordToPerTopicProb: Map[String, Array[Double]] = Map(("word_1" -> Array.fill(4)(0.05))) + val wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]] = sparkContext.broadcast(wordToPerTopicProb) + val topicCount = 4 + + val precisionUtility = FloatPointPrecisionUtility64 + val documentProbabilities = Seq(0.05d, 0.05d, 0.05d, 0.05d) + + val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) + + val score = scoreFunction.score(precisionUtility)(documentProbabilities, "word_1") + + score shouldBe 0.010000000000000002d + + } + + it should "return score of type Double with document probabilities of type Float" in { + + val wordToPerTopicProb: Map[String, Array[Double]] = Map(("word_1" -> Array.fill(4)(0.05))) + val wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]] = sparkContext.broadcast(wordToPerTopicProb) + val topicCount = 4 + + val precisionUtility = FloatPointPrecisionUtility32 + val documentProbabilities = Seq(0.05f, 0.05f, 0.05f, 0.05f) + + val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) + + val score = scoreFunction.score(precisionUtility)(documentProbabilities, "word_1") + + score shouldBe 0.010000000149011612d + + } + + it should "return score -1 when word doesn't exists" in { + val wordToPerTopicProb: Map[String, Array[Double]] = Map(("word_1" -> Array.fill(4)(0.05))) + val wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]] = sparkContext.broadcast(wordToPerTopicProb) + val topicCount = 4 + + val precisionUtility = FloatPointPrecisionUtility32 + val documentProbabilities = Seq(0.05f, 0.05f, 0.05f, 0.05f) + + val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC) + + val score = scoreFunction.score(precisionUtility)(documentProbabilities, "word_error") + + score shouldBe -1d + } + +} diff --git a/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala index 40c5189b..90b26eb4 100644 --- a/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala +++ b/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala @@ -19,11 +19,12 @@ package org.apache.spot.dns import org.apache.log4j.{Level, LogManager} -import org.apache.spark.sql.{Row, SQLContext} import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.{Row, SQLContext} import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig import org.apache.spot.dns.DNSSchema._ import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.apache.spot.utilities.FloatPointPrecisionUtility32 import org.scalatest.Matchers @@ -52,8 +53,23 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with ldaAlpha = 1.02, ldaBeta = 1.001) + val testConfigFloatConversion = SuspiciousConnectsConfig(analysis = "dns", + inputPath = "", + feedbackFile = "", + duplicationFactor = 1, + topicCount = 20, + hdfsScoredConnect = "", + threshold = 1.0d, + maxResults = 1000, + outputDelimiter = "\t", + ldaPRGSeed = None, + ldaMaxiterations = 20, + ldaAlpha = 1.02, + ldaBeta = 1.001, + precisionUtility = FloatPointPrecisionUtility32) - "dns supicious connects analysis" should "estimate correct probabilities in toy data with framelength anomaly" in { + + "dns suspicious connects analysis" should "estimate correct probabilities in toy data with framelength anomaly" in { val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") logger.setLevel(Level.WARN) @@ -74,7 +90,7 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with } - "dns supicious connects analysis" should "estimate correct probabilities in toy data with subdomain length anomaly" in { + it should "estimate correct probabilities in toy data with subdomain length anomaly" in { val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") logger.setLevel(Level.WARN) @@ -111,6 +127,65 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Math.abs(typicalScores(3) - 0.8d) should be <= 0.01d } + it should "estimate correct probabilities in toy data with framelength anomaly converting probabilities to Float " + + "for transportation and converting back to Double" in { + + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + + val anomalousRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 1, "172.16.9.132", "122.2o7.turner.com", "0x00000001", 1, 0) + val typicalRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 168, "172.16.9.132", "122.2o7.turner.com", "0x00000001", 1, 0) + val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord)) + val scoredData = DNSSuspiciousConnectsAnalysis.scoreDNSRecords(data, testConfigFloatConversion, sparkContext, sqlContext, logger) + val anomalyScore = scoredData.filter(scoredData(FrameLength) === 1).first().getAs[Double](Score) + val typicalScores = scoredData.filter(scoredData(FrameLength) === 168).collect().map(_.getAs[Double](Score)) + + Math.abs(anomalyScore - 0.2d) should be <= 0.01d + typicalScores.length shouldBe 4 + Math.abs(typicalScores(0) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(1) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(2) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(3) - 0.8d) should be <= 0.01d + } + + + it should "estimate correct probabilities in toy data with subdomain length anomaly converting probabilities to " + + "Float for transportation and converting back to Double" in { + + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + + val anomalousRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", + 1463735425L, + 168, + "172.16.9.132", + "1111111111111111111111111111111111111111111111111111111111111.tinker.turner.com", + "0x00000001", + 1, + 0) + val typicalRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", + 1463735425L, + 168, + "172.16.9.132", + "tinker.turner.com", + "0x00000001", + 1, + 0) + val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord)) + val scoredData = DNSSuspiciousConnectsAnalysis.scoreDNSRecords(data, testConfigFloatConversion, sparkContext, sqlContext, logger) + val anomalyScore = scoredData. + filter(scoredData(QueryName) === "1111111111111111111111111111111111111111111111111111111111111.tinker.turner.com"). + first(). + getAs[Double](Score) + val typicalScores = scoredData.filter(scoredData(QueryName) === "tinker.turner.com").collect().map(_.getAs[Double](Score)) + + Math.abs(anomalyScore - 0.2d) should be <= 0.01d + typicalScores.length shouldBe 4 + Math.abs(typicalScores(0) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(1) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(2) - 0.8d) should be <= 0.01d + Math.abs(typicalScores(3) - 0.8d) should be <= 0.01d + } "filterAndSelectCleanDNSRecords" should "return data set without garbage" in { diff --git a/spot-ml/src/test/scala/org/apache/spot/lda/SpotLDAWrapperTest.scala b/spot-ml/src/test/scala/org/apache/spot/lda/SpotLDAWrapperTest.scala new file mode 100644 index 00000000..e701bb90 --- /dev/null +++ b/spot-ml/src/test/scala/org/apache/spot/lda/SpotLDAWrapperTest.scala @@ -0,0 +1,244 @@ +/* + * 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.lda + +import org.apache.log4j.{Level, LogManager} +import org.apache.spark.mllib.linalg.{Matrices, Vector, Vectors} +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.{DataFrame, Row} +import org.apache.spot.lda.SpotLDAWrapper._ +import org.apache.spot.lda.SpotLDAWrapperSchema._ +import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.apache.spot.utilities.{FloatPointPrecisionUtility32, FloatPointPrecisionUtility64} +import org.scalatest.Matchers + +import scala.collection.immutable.Map + +class SpotLDAWrapperTest extends TestingSparkContextFlatSpec with Matchers { + + val ldaAlpha = 1.02 + val ldaBeta = 1.001 + val ldaMaxIterations = 20 + + "SparkLDA" should "handle an extremely unbalanced two word doc" in { + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + + val catFancy = SpotLDAInput("pets", "cat", 1) + val dogWorld = SpotLDAInput("pets", "dog", 999) + + val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) + val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, + ldaMaxIterations, FloatPointPrecisionUtility64) + + val topicMixDF = out.docToTopicMix + + val topicMix = + topicMixDF.filter(topicMixDF(DocumentName) === "pets").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Double]].toArray + val catTopics = out.wordResults("cat") + val dogTopics = out.wordResults("dog") + + Math.abs(topicMix(0) * catTopics(0) + topicMix(1) * catTopics(1)) should be < 0.01 + Math.abs(0.999 - (topicMix(0) * dogTopics(0) + topicMix(1) * dogTopics(1))) should be < 0.01 + } + + it should "handle an extremely unbalanced two word doc with doc probabilities as Float" in { + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + + val catFancy = SpotLDAInput("pets", "cat", 1) + val dogWorld = SpotLDAInput("pets", "dog", 999) + + val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) + val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, + ldaMaxIterations, FloatPointPrecisionUtility32) + + val topicMixDF = out.docToTopicMix + + val topicMix = + topicMixDF.filter(topicMixDF(DocumentName) === "pets").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Float]].toArray + val catTopics = out.wordResults("cat") + val dogTopics = out.wordResults("dog") + + Math.abs(topicMix(0).toDouble * catTopics(0) + topicMix(1).toDouble * catTopics(1)) should be < 0.01 + Math.abs(0.999 - (topicMix(0).toDouble * dogTopics(0) + topicMix(1).toDouble * dogTopics(1))) should be < 0.01 + } + + it should "handle distinct docs on distinct words" in { + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + val catFancy = SpotLDAInput("cat fancy", "cat", 1) + val dogWorld = SpotLDAInput("dog world", "dog", 1) + + val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) + val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, + ldaMaxIterations, FloatPointPrecisionUtility64) + + val topicMixDF = out.docToTopicMix + val dogTopicMix: Array[Double] = + topicMixDF.filter(topicMixDF(DocumentName) === "dog world").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Double]].toArray + + val catTopicMix: Array[Double] = + topicMixDF.filter(topicMixDF(DocumentName) === "cat fancy").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Double]].toArray + + val catTopics = out.wordResults("cat") + val dogTopics = out.wordResults("dog") + + Math.abs(1 - (catTopicMix(0) * catTopics(0) + catTopicMix(1) * catTopics(1))) should be < 0.01 + Math.abs(1 - (dogTopicMix(0) * dogTopics(0) + dogTopicMix(1) * dogTopics(1))) should be < 0.01 + } + + it should "handle distinct docs on distinct words with doc probabilities as Float" in { + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + val catFancy = SpotLDAInput("cat fancy", "cat", 1) + val dogWorld = SpotLDAInput("dog world", "dog", 1) + + val data = sparkContext.parallelize(Seq(catFancy, dogWorld)) + val out = SpotLDAWrapper.runLDA(sparkContext, sqlContext, data, 2, logger, Some(0xdeadbeef), ldaAlpha, ldaBeta, + ldaMaxIterations, FloatPointPrecisionUtility32) + + val topicMixDF = out.docToTopicMix + val dogTopicMix: Array[Float] = + topicMixDF.filter(topicMixDF(DocumentName) === "dog world").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Float]].toArray + + val catTopicMix: Array[Float] = + topicMixDF.filter(topicMixDF(DocumentName) === "cat fancy").select(TopicProbabilityMix).first().toSeq(0) + .asInstanceOf[Seq[Float]].toArray + + val catTopics = out.wordResults("cat") + val dogTopics = out.wordResults("dog") + + Math.abs(1 - (catTopicMix(0) * catTopics(0) + catTopicMix(1) * catTopics(1))) should be < 0.01 + Math.abs(1 - (dogTopicMix(0) * dogTopics(0) + dogTopicMix(1) * dogTopics(1))) should be < 0.01 + } + + "formatSparkLDAInput" should "return input in RDD[(Long, Vector)] (collected as Array for testing) format. The index " + + "is the docID, values are the vectors of word occurrences in that doc" in { + + + val documentWordData = sparkContext.parallelize(Seq(SpotLDAInput("192.168.1.1", "333333_7.0_0.0_1.0", 8), + SpotLDAInput("10.10.98.123", "1111111_6.0_3.0_5.0", 4), + SpotLDAInput("66.23.45.11", "-1_43_7.0_2.0_6.0", 2), + SpotLDAInput("192.168.1.1", "-1_80_6.0_1.0_1.0", 5))) + + val wordDictionary = Map("333333_7.0_0.0_1.0" -> 0, "1111111_6.0_3.0_5.0" -> 1, "-1_43_7.0_2.0_6.0" -> 2, + "-1_80_6.0_1.0_1.0" -> 3) + + val documentDictionary: DataFrame = sqlContext.createDataFrame(documentWordData + .map({ case SpotLDAInput(doc, word, count) => doc }) + .distinct + .zipWithIndex.map({ case (d, c) => Row(d, c) }), StructType(List(DocumentNameField, DocumentNumberField))) + + + val sparkLDAInput: RDD[(Long, Vector)] = SpotLDAWrapper.formatSparkLDAInput(documentWordData, + documentDictionary, wordDictionary, sqlContext) + val sparkLDAInArr: Array[(Long, Vector)] = sparkLDAInput.collect() + + sparkLDAInArr shouldBe Array((0, Vectors.sparse(4, Array(0, 3), Array(8.0, 5.0))), (2, Vectors.sparse(4, Array + (2), Array(2.0))), (1, Vectors.sparse(4, Array(1), Array(4.0)))) + } + + "formatSparkLDADocTopicOutput" should "return RDD[(String,Array(Double))] after converting doc results from vector " + + "using PrecisionUtilityDouble: convert docID back to string, convert vector of probabilities to array" in { + + val documentWordData = sparkContext.parallelize(Seq(SpotLDAInput("192.168.1.1", "333333_7.0_0.0_1.0", 8), + SpotLDAInput("10.10.98.123", "1111111_6.0_3.0_5.0", 4), + SpotLDAInput("66.23.45.11", "-1_43_7.0_2.0_6.0", 2), + SpotLDAInput("192.168.1.1", "-1_80_6.0_1.0_1.0", 5))) + + val documentDictionary: DataFrame = sqlContext.createDataFrame(documentWordData + .map({ case SpotLDAInput(doc, word, count) => doc }) + .distinct + .zipWithIndex.map({ case (d, c) => Row(d, c) }), StructType(List(DocumentNameField, DocumentNumberField))) + + val docTopicDist: RDD[(Long, Vector)] = sparkContext.parallelize( + Array((0.toLong, Vectors.dense(0.15, 0.3, 0.5, 0.05)), + (1.toLong, Vectors.dense(0.25, 0.15, 0.4, 0.2)), + (2.toLong, Vectors.dense(0.4, 0.1, 0.3, 0.2)))) + + val sparkDocRes: DataFrame = formatSparkLDADocTopicOutput(docTopicDist, documentDictionary, sqlContext, + FloatPointPrecisionUtility64) + + val documents = sparkDocRes.select(DocumentName).map(documentName => documentName.toString.replaceAll("\\[", "") + .replaceAll("\\]", "")).collect() + + val documentProbabilities = sparkDocRes.select(TopicProbabilityMix).first.toSeq(0).asInstanceOf[Seq[Double]] + + documents(0) should be("10.10.98.123") + documents(1) should be("192.168.1.1") + documents(2) should be("66.23.45.11") + + documentProbabilities(0) shouldBe a[java.lang.Double] + + } + + it should "return RDD[(String,Array(Float))] after converting doc results from vector " + + "using PrecisionUtilityFloat: convert docID back to string, convert vector of probabilities to array" in { + + val documentWordData = sparkContext.parallelize(Seq(SpotLDAInput("192.168.1.1", "333333_7.0_0.0_1.0", 8), + SpotLDAInput("10.10.98.123", "1111111_6.0_3.0_5.0", 4), + SpotLDAInput("66.23.45.11", "-1_43_7.0_2.0_6.0", 2), + SpotLDAInput("192.168.1.1", "-1_80_6.0_1.0_1.0", 5))) + + val documentDictionary: DataFrame = sqlContext.createDataFrame(documentWordData + .map({ case SpotLDAInput(doc, word, count) => doc }) + .distinct + .zipWithIndex.map({ case (d, c) => Row(d, c) }), StructType(List(DocumentNameField, DocumentNumberField))) + + val docTopicDist: RDD[(Long, Vector)] = sparkContext.parallelize( + Array((0.toLong, Vectors.dense(0.15, 0.3, 0.5, 0.05)), + (1.toLong, Vectors.dense(0.25, 0.15, 0.4, 0.2)), + (2.toLong, Vectors.dense(0.4, 0.1, 0.3, 0.2)))) + + val sparkDocRes: DataFrame = formatSparkLDADocTopicOutput(docTopicDist, documentDictionary, sqlContext, + FloatPointPrecisionUtility32) + + val documents = sparkDocRes.select(DocumentName).map(documentName => documentName.toString.replaceAll("\\[", "") + .replaceAll("\\]", "")).collect() + + val documentProbabilities = sparkDocRes.select(TopicProbabilityMix).first.toSeq(0).asInstanceOf[Seq[Float]] + + documents(0) should be("10.10.98.123") + documents(1) should be("192.168.1.1") + documents(2) should be("66.23.45.11") + + documentProbabilities(0) shouldBe a[java.lang.Float] + } + + "formatSparkLDAWordOutput" should "return Map[Int,String] after converting word matrix to sequence, wordIDs back " + + "to strings, and sequence of probabilities to array" in { + val testMat = Matrices.dense(4, 4, Array(0.5, 0.2, 0.05, 0.25, 0.25, 0.1, 0.15, 0.5, 0.1, 0.4, 0.25, 0.25, 0.7, 0.2, 0.02, 0.08)) + + val wordDictionary = Map("-1_23.0_7.0_7.0_4.0" -> 3, "23.0_7.0_7.0_4.0" -> 0, "333333.0_7.0_7.0_4.0" -> 2, "80.0_7.0_7.0_4.0" -> 1) + val revWordMap: Map[Int, String] = wordDictionary.map(_.swap) + + val sparkWordRes = formatSparkLDAWordOutput(testMat, revWordMap) + + sparkWordRes should contain key ("23.0_7.0_7.0_4.0") + sparkWordRes should contain key ("80.0_7.0_7.0_4.0") + sparkWordRes should contain key ("333333.0_7.0_7.0_4.0") + sparkWordRes should contain key ("-1_23.0_7.0_7.0_4.0") + } +} \ No newline at end of file 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 29e3363f..d735a19f 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 @@ -23,6 +23,7 @@ import org.apache.spark.sql.{DataFrame, Row, SQLContext} import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig import org.apache.spot.netflow.FlowSchema._ import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.apache.spot.utilities.FloatPointPrecisionUtility32 import org.scalatest.Matchers case class FlowRecord(treceived: String, @@ -38,8 +39,8 @@ case class FlowRecord(treceived: String, sport: Int, dport: Int, proto: String, - ipkt: Int, - ibyt: Int, + ipkt: Long, + ibyt: Long, opkt: Int, obyt: Int) @@ -60,6 +61,21 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit ldaAlpha = 1.02, ldaBeta = 1.001) + val testingConfigFloatConversion = 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, + precisionUtility = FloatPointPrecisionUtility32) + "netflow suspicious connects" should "correctly identify time-of-day anomalies" in { val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") @@ -80,6 +96,44 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit logger) + val anomalyScore = scoredData.filter(scoredData(Hour) === 0).first().getAs[Double](Score) + val typicalScores = scoredData.filter(scoredData(Hour) === 13).collect().map(_.getAs[Double](Score)) + + Math.abs(anomalyScore - 0.1d) should be < 0.01 + typicalScores.length shouldBe 9 + Math.abs(typicalScores(0) - 0.9d) should be < 0.01 + Math.abs(typicalScores(1) - 0.9d) should be < 0.01 + Math.abs(typicalScores(2) - 0.9d) should be < 0.01 + Math.abs(typicalScores(3) - 0.9d) should be < 0.01 + Math.abs(typicalScores(4) - 0.9d) should be < 0.01 + Math.abs(typicalScores(5) - 0.9d) should be < 0.01 + Math.abs(typicalScores(6) - 0.9d) should be < 0.01 + Math.abs(typicalScores(7) - 0.9d) should be < 0.01 + Math.abs(typicalScores(8) - 0.9d) should be < 0.01 + + + } + + it should "correctly identify time-of-day anomalies converting probabilities to Float for transportation and " + + "converting back to Double" in { + + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.OFF) + + val anomalousRecord = FlowRecord("2016-05-05 00:11:01", 2016, 5, 5, 0, 0, 1, 0.972f, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39, 12522, 0, 0) + val typicalRecord = FlowRecord("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972f, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39, 12522, 0, 0) + + + val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, + typicalRecord, typicalRecord, typicalRecord, typicalRecord)) + + + val scoredData: DataFrame = FlowSuspiciousConnectsAnalysis.detectFlowAnomalies(data, + testingConfigFloatConversion, + sparkContext, + sqlContext, + logger) + val anomalyScore = scoredData.filter(scoredData(Hour) === 0).first().getAs[Double](Score) val typicalScores = scoredData.filter(scoredData(Hour) === 13).collect().map(_.getAs[Double](Score)) @@ -98,6 +152,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit } + "filterAndSelectCleanFlowRecords" should "return data set without garbage" in { val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis diff --git a/spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala index 8b0b9c6a..ebcd726e 100644 --- a/spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala +++ b/spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala @@ -18,37 +18,37 @@ package org.apache.spot.proxy import org.apache.log4j.{Level, LogManager} -import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig -import org.apache.spark.sql.{Row, SQLContext} import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.{Row, SQLContext} +import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig import org.apache.spot.proxy.ProxySchema._ import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.apache.spot.utilities.FloatPointPrecisionUtility32 import org.scalatest.Matchers -case class ProxyInput(p_date:String, - p_time:String, - clientip:String, - host:String, - reqmethod:String, - useragent:String, - resconttype:String, - duration:Int, - username:String, - webcat:String, - referer:String, - respcode:String, - uriport:Int, - uripath:String, - uriquery:String, - serverip:String, - scbytes:Int, - csbytes:Int, - fulluri:String) +case class ProxyInput(p_date: String, + p_time: String, + clientip: String, + host: String, + reqmethod: String, + useragent: String, + resconttype: String, + duration: Int, + username: String, + webcat: String, + referer: String, + respcode: String, + uriport: Int, + uripath: String, + uriquery: String, + serverip: String, + scbytes: Int, + csbytes: Int, + fulluri: String) class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Matchers { - val testConfigProxy = SuspiciousConnectsConfig(analysis = "proxy", inputPath = "", feedbackFile = "", @@ -63,26 +63,29 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi ldaAlpha = 1.02, ldaBeta = 1.001) + val testConfigProxyFloatConversion = SuspiciousConnectsConfig(analysis = "proxy", + inputPath = "", + feedbackFile = "", + duplicationFactor = 1, + topicCount = 20, + hdfsScoredConnect = "", + threshold = 1.0d, + maxResults = 1000, + outputDelimiter = "\t", + ldaPRGSeed = None, + ldaMaxiterations = 20, + ldaAlpha = 1.02, + ldaBeta = 1.001, + precisionUtility = FloatPointPrecisionUtility32) - "proxy supicious connects analysis" should "estimate correct probabilities in toy data with top domain anomaly" in { + + "proxy suspicious connects analysis" should "estimate correct probabilities in toy data with top domain anomaly" in { val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") logger.setLevel(Level.WARN) - val anomalousRecord = ProxyInput("2016-10-03", "04:57:36", "127.0.0.1", "intel.com", "PUT", - "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36", - "text/plain", 230, "-", "Technology/Internet", "http://www.spoonflower.com/tags/color", "202", 80, - "/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle", - "-", "127.0.0.1", 338, 647, - "maw.bronto.com/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle") - - val typicalRecord = ProxyInput("2016-10-03", "04:57:36", "127.0.0.1", "maw.bronto.com", "PUT", - "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36", - "text/plain", 230, "-", "Technology/Internet", "http://www.spoonflower.com/tags/color", "202", 80, - "/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle", - "-", "127.0.0.1", 338, 647, - "maw.bronto.com/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle") + val (anomalousRecord, typicalRecord) = anomalousAndTypicalRecords() val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord)) @@ -93,23 +96,55 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi logger) - - val anomalyScore = scoredData.filter(scoredData(Host) === "intel.com").first().getAs[Double](Score) + val anomalyScore = scoredData.filter(scoredData(Host) === "intel.com").first().getAs[Double](Score) val typicalScores = scoredData.filter(scoredData(Host) === "maw.bronto.com").collect().map(_.getAs[Double](Score)) - Math.abs(anomalyScore - 0.1d) should be <= 0.01d + Math.abs(anomalyScore - 0.1d) should be <= 0.01d typicalScores.length shouldBe 9 - Math.abs(typicalScores(0) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(1) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(2) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(3) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(4) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(5) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(6) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(7) - 0.9d) should be <= 0.01d - Math.abs(typicalScores(8) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(0) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(1) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(2) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(3) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(4) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(5) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(6) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(7) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(8) - 0.9d) should be <= 0.01d } + "proxy suspicious connects analysis" should "estimate correct probabilities in toy data with top domain anomaly " + + "converting probabilities to Float for transportation and converting back to Double" in { + + val logger = LogManager.getLogger("SuspiciousConnectsAnalysis") + logger.setLevel(Level.WARN) + + val (anomalousRecord, typicalRecord) = anomalousAndTypicalRecords() + + + val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord, + typicalRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord)) + + val scoredData = ProxySuspiciousConnectsAnalysis.detectProxyAnomalies(data, testConfigProxyFloatConversion, + sparkContext, + sqlContext, + logger) + + + val anomalyScore = scoredData.filter(scoredData(Host) === "intel.com").first().getAs[Double](Score) + val typicalScores = scoredData.filter(scoredData(Host) === "maw.bronto.com").collect().map(_.getAs[Double](Score)) + + Math.abs(anomalyScore - 0.1d) should be <= 0.01d + typicalScores.length shouldBe 9 + Math.abs(typicalScores(0) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(1) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(2) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(3) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(4) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(5) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(6) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(7) - 0.9d) should be <= 0.01d + Math.abs(typicalScores(8) - 0.9d) should be <= 0.01d + } "filterAndSelectCleanProxyRecords" should "return data without garbage" in { @@ -150,23 +185,41 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi corruptProxyRecords.schema.size should be(21) } + def anomalousAndTypicalRecords(): (ProxyInput, ProxyInput) = { + val anomalousRecord = ProxyInput("2016-10-03", "04:57:36", "127.0.0.1", "intel.com", "PUT", + "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36", + "text/plain", 230, "-", "Technology/Internet", "http://www.spoonflower.com/tags/color", "202", 80, + "/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle", + "-", "127.0.0.1", 338, 647, + "maw.bronto.com/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle") + + val typicalRecord = ProxyInput("2016-10-03", "04:57:36", "127.0.0.1", "maw.bronto.com", "PUT", + "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36", + "text/plain", 230, "-", "Technology/Internet", "http://www.spoonflower.com/tags/color", "202", 80, + "/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle", + "-", "127.0.0.1", 338, 647, + "maw.bronto.com/sites/c37i4q22szvir8ga3m8mtxaft7gwnm5fio8hfxo35mu81absi1/carts/4b3a313d-50f6-4117-8ffd-4e804fd354ef/fiddle") + + (anomalousRecord, typicalRecord) + } + def testProxyRecords = new { val sqlContext = new SQLContext(sparkContext) val inputProxyRecordsRDD = sparkContext.parallelize(wrapRefArray(Array( - Seq(null,"00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu..."), - Seq("2016-10-03",null,"10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu..."), - Seq("2016-10-03","00:09:13",null,"cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu..."), - Seq("2016-10-03","00:09:13","10.239.160.152",null,"GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu..."), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,null), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...")) + Seq(null, "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu..."), + Seq("2016-10-03", null, "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu..."), + Seq("2016-10-03", "00:09:13", null, "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu..."), + Seq("2016-10-03", "00:09:13", "10.239.160.152", null, "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu..."), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, null), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...")) .map(row => Row.fromSeq(row)))) val inputProxyRecordsSchema = StructType( @@ -193,16 +246,16 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi val inputProxyRecordsDF = sqlContext.createDataFrame(inputProxyRecordsRDD, inputProxyRecordsSchema) val scoredProxyRecordsRDD = sparkContext.parallelize(wrapRefArray(Array( - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...", "a word", -1d), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...", "a word", 1d), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...", "a word", 0.0000005), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...", "a word", 0.05), - Seq("2016-10-03","00:09:13","10.239.160.152","cn.archive.ubuntu...","GET","Debian APT-HTTP/...","text/html",448,"-", - "-","-","404","80","/ubuntu/dists/tru...","-","10.239.4.160",2864,218,"cn.archive.ubuntu...", "a word", 0.0001) + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...", "a word", -1d), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...", "a word", 1d), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...", "a word", 0.0000005), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...", "a word", 0.05), + Seq("2016-10-03", "00:09:13", "10.239.160.152", "cn.archive.ubuntu...", "GET", "Debian APT-HTTP/...", "text/html", 448, "-", + "-", "-", "404", "80", "/ubuntu/dists/tru...", "-", "10.239.4.160", 2864, 218, "cn.archive.ubuntu...", "a word", 0.0001) ).map(row => Row.fromSeq(row)))) val scoredProxyRecordsSchema = StructType( diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.scala index 0dadfc0c..692b1cda 100644 --- a/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.scala +++ b/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.scala @@ -18,12 +18,13 @@ package org.apache.spot.utilities import org.apache.spot.testutils.TestingSparkContextFlatSpec -import org.scalatest.{FunSuite, Matchers} +import org.apache.spot.utilities import org.apache.spot.utilities.DomainProcessor._ +import org.scalatest.Matchers class DomainProcessorTest extends TestingSparkContextFlatSpec with Matchers { - val countryCodesSet = CountryCodes.CountryCodes + val countryCodesSet = utilities.CountryCodes.CountryCodes "extractDomain" should "return domain when provided a url with top-level domain and country code" in { diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/FloatPointPrecisionUtility32Test.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/FloatPointPrecisionUtility32Test.scala new file mode 100644 index 00000000..de66876a --- /dev/null +++ b/spot-ml/src/test/scala/org/apache/spot/utilities/FloatPointPrecisionUtility32Test.scala @@ -0,0 +1,49 @@ +package org.apache.spot.utilities + +import org.apache.spark.sql.types._ +import org.apache.spot.lda.SpotLDAWrapperSchema.{DocumentName, TopicProbabilityMix} +import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.scalatest.Matchers + +/** + * Created by rabarona on 5/17/17. + */ +class FloatPointPrecisionUtility32Test extends TestingSparkContextFlatSpec with Matchers { + + "toTargetType" should "just return value converted to Float" in { + val testValue: Double = 5d + + val result = FloatPointPrecisionUtility32.toTargetType(testValue) + + result shouldBe 5f + result shouldBe a[java.lang.Float] + + } + + "toDoubles" should "convert a Seq of Float to Seq of Double" in { + + val testSeq: Seq[Float] = Seq(1f, 2f, 3f) + + val result: Seq[Double] = FloatPointPrecisionUtility32.toDoubles(testSeq) + + result(0) shouldBe a[java.lang.Double] + result(1) shouldBe a[java.lang.Double] + result(2) shouldBe a[java.lang.Double] + } + + "castColumn" should "return a data frame with a schema modified, Seq[Float] instead of Seq[Double]" in { + + val testDataFrame = sqlContext.createDataFrame(Seq(("doc1", Array(1d, 2d)), ("doc2", Array(2d, 3d)))) + .withColumnRenamed("_1", DocumentName).withColumnRenamed("_2", TopicProbabilityMix) + + val result = FloatPointPrecisionUtility32.castColumn(testDataFrame, TopicProbabilityMix) + result.count + + val schema = StructType( + Array(StructField(DocumentName, StringType, true), + StructField(TopicProbabilityMix, ArrayType(FloatType, false), true))) + + result.schema shouldBe schema + + } +} diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/FloatingPointUtility64.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/FloatingPointUtility64.scala new file mode 100644 index 00000000..2006fde9 --- /dev/null +++ b/spot-ml/src/test/scala/org/apache/spot/utilities/FloatingPointUtility64.scala @@ -0,0 +1,48 @@ +package org.apache.spot.utilities + +import org.apache.spark.sql.types._ +import org.apache.spot.lda.SpotLDAWrapperSchema.{DocumentName, TopicProbabilityMix} +import org.apache.spot.testutils.TestingSparkContextFlatSpec +import org.scalatest.Matchers + +/** + * Created by rabarona on 5/17/17. + */ +class FloatingPointUtility64 extends TestingSparkContextFlatSpec with Matchers { + + "toTargetType" should "just return the same value with the same type" in { + val testValue: Double = 5d + + val result = FloatPointPrecisionUtility64.toTargetType(testValue) + + result shouldBe testValue + result shouldBe a[java.lang.Double] + + } + + "toDoubles" should "return an array of the same type" in { + + val testSeq: Seq[Double] = Seq(1d, 2d, 3d) + + val result: Seq[Double] = FloatPointPrecisionUtility64.toDoubles(testSeq) + + result shouldBe a[Seq[Double]] + result.length shouldBe 3 + } + + "castColumn" should "return the exact same data frame" in { + + val testDataFrame = sqlContext.createDataFrame(Seq(("doc1", Array(1d, 2d)), ("doc2", Array(2d, 3d)))) + .withColumnRenamed("_1", DocumentName).withColumnRenamed("_2", TopicProbabilityMix) + + val result = FloatPointPrecisionUtility64.castColumn(testDataFrame, TopicProbabilityMix) + + val schema = StructType( + Array(StructField(DocumentName, StringType, true), + StructField(TopicProbabilityMix, ArrayType(DoubleType, false), true))) + + result shouldBe testDataFrame + result.schema shouldBe schema + } + +} diff --git a/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/QuantilesTest.scala similarity index 99% rename from spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala rename to spot-ml/src/test/scala/org/apache/spot/utilities/QuantilesTest.scala index 54fec85f..bf71abea 100644 --- a/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala +++ b/spot-ml/src/test/scala/org/apache/spot/utilities/QuantilesTest.scala @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.spot +package org.apache.spot.utilities import org.apache.spark.rdd.RDD import org.apache.spot.testutils.TestingSparkContextFlatSpec diff --git a/spot-setup/spot.conf b/spot-setup/spot.conf index 65fbd744..6fe30588 100755 --- a/spot-setup/spot.conf +++ b/spot-setup/spot.conf @@ -37,7 +37,9 @@ SPK_DRIVER_MAX_RESULTS='' SPK_EXEC_CORES='' SPK_DRIVER_MEM_OVERHEAD='' SPK_EXEC_MEM_OVERHEAD='' -TOL='1e-6' +SPK_AUTO_BRDCST_JOIN_THR='10485760' +PRECISION='64' +TOL='1e-6' TOPIC_COUNT=20 DUPFACTOR=1000