You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by nl...@apache.org on 2017/06/15 21:56:34 UTC
[04/13] incubator-spot git commit: Added scaling option to convert
document probabilities from Double to Float and reduce the payload when
joining original data and LDA results (document probabilities). Added
parameters for using Doubles (64 bit) or Floa
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/ProbabilityConverter.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/ProbabilityConverter.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/ProbabilityConverter.scala
new file mode 100644
index 0000000..67d8d6b
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/ProbabilityConverter.scala
@@ -0,0 +1,110 @@
+/*
+ * 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.transformation
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions._
+
+/**
+ * ProbabilityConverter will transform document probability from Double to Float if scaling option is set to 32 bit,
+ * if default or 64 bit is selected, it will just return the same Double probability.
+ *
+ * This abstract class permits the execution of a single path during the entire analysis. Instead of checking what
+ * type should the data be converted, each implementation will know what type should be converted into and converted
+ * back.
+ *
+ */
+
+abstract class ProbabilityConverter extends Serializable {
+
+ type ScalingType
+
+ /**
+ * Converts a probability into the scaling type; it can be Float or Double.
+ * For the Double implementation it will just return the same value without any transformation.
+ *
+ * @param probability a probability to convert from Double to scaling type.
+ * @return
+ */
+ def convertProbability(probability: Double): ScalingType
+
+ /**
+ * Converts back a set of probabilities from the scaling type into Double.
+ *
+ * @param scaledProbabilities
+ * @return
+ */
+ def convertBackSetOfProbabilities(scaledProbabilities: Seq[ScalingType]): Array[Double]
+
+ /**
+ * Converts a column for a given DataFrame from Double to the scaling type. If the scaling type is Double, it will
+ * just return the same DataFrame without any transformation.
+ *
+ * @param dataFrame a DataFrame containing a column to be converted from Double to the scaling type
+ * @param columnName the name of the column to be converted.
+ * @return
+ */
+ def convertDataFrameColumn(dataFrame: DataFrame, columnName: String): DataFrame
+
+}
+
+/**
+ * ProbabilityConverter implementation for Float.
+ * Will convert probabilities from Double to Float and back to Double to make the workload half the size that it'd
+ * be with Double. This conversion will permit to broadcast document probabilities if it falls in the
+ * autoBroadcastJoinThreshold - defined by users - and accelerate the join process between document probabilities
+ * and the entire data set.
+ */
+object ProbabilityConverterFloat extends ProbabilityConverter {
+
+ type ScalingType = Float
+ val convertUDF = udf((probabilities: Seq[Double]) => {
+ for (probability <- probabilities) yield convertProbability(probability)
+ })
+
+ def convertProbability(probability: Double): Float = probability.toFloat
+
+ def convertBackSetOfProbabilities(scaledProbabilities: Seq[Float]): Array[Double] =
+ for (scaledProbability <- scaledProbabilities.toArray) yield scaledProbability.toDouble
+
+ override def convertDataFrameColumn(dataFrame: DataFrame, columnName: String): DataFrame = {
+ dataFrame.withColumn("temp_column", convertUDF(dataFrame(columnName)))
+ .drop(columnName)
+ .withColumnRenamed("temp_column", columnName)
+ }
+}
+
+/**
+ * ProbabilityConverter implementation for Double.
+ * Users that don't want to reduce the workload, can continue working with Doubles. This implementation will receive
+ * and send the same value, it won't do any transformation. The reason for this to exists is to avoid multiple code
+ * paths depending on the user wanting to scale or not.
+ */
+object ProbabilityConverterDouble extends ProbabilityConverter {
+
+ type ScalingType = Double
+
+ def convertProbability(probability: Double): Double = probability
+
+ // Since Double is the default data type, this code is going to return the same array without any transformation.
+ def convertBackSetOfProbabilities(scaledProbability: Seq[Double]): Array[Double] = scaledProbability.toArray
+
+ // Since Double is the default data type, this code won't actually do any calculation but instead just return the
+ // same DataFrame.
+ override def convertDataFrameColumn(dataFrame: DataFrame, columnName: String): DataFrame = dataFrame
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/Quantiles.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/Quantiles.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/Quantiles.scala
new file mode 100644
index 0000000..20ccd40
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/Quantiles.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.transformation
+
+import org.apache.spark.rdd.RDD
+
+import scala.math._
+
+/**
+ * Contains routines for the distributed calculation of quantiles and the empirical cumulative distribution function.
+ */
+
+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 deciles of a distribution.
+ *
+ * @param data RDD[Double] Incoming data.
+ * @return Array[Double]. The deciles of the distribution.
+ */
+
+ def computeDeciles(data: RDD[Double]): Array[Double] = computeQuantiles(data, DECILES)
+
+ /**
+ * 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.
+ *
+ * @param data Incoming dataset.
+ * @param quantiles Sorted array of doubles in the range from 0.0 to 1.0.
+ * @return The quantiles of the data, that is, if t is the ith entry of the quantiles array, and x is the
+ * ith entry of returned quantiles array, then x is the minimum value in the input dataset so that
+ * a >= t fraction of the mass of the input is <= x.
+ * IE. x is the least x so that Pr ( X <= x ) >= t
+ * In the case of an empty input dataset, the quantile cutoffs returned are all positive infinity.
+ */
+
+ def computeQuantiles(data: RDD[Double], quantiles: Array[Double]): Array[Double] = {
+
+ /*
+ * Calculation is based on the fact that the quantile of a threshold t is the minimum x in the dataset so
+ * that the empirical cumulative distribution at x is >= t.
+ */
+
+ def addDataPointToKnownCutoffs(cutoffs: Array[Double], newValueCDFPair: (Double, Double)): Array[Double] = {
+ val newPoint = newValueCDFPair._1
+ val cdfAtNewPoint = newValueCDFPair._2
+ quantiles.zip(cutoffs).map({ case (threshold, oldCutoff) =>
+ if (cdfAtNewPoint >= threshold && newPoint <= oldCutoff) newPoint else oldCutoff
+ })
+ }
+
+ def mergeCutoffs(cuts1: Array[Double], cuts2: Array[Double]) = cuts1.zip(cuts2).map({ case (x, y) => min(x, y) })
+
+ // Initial cutoffs are the trivial "positive infinity" cutoffs. These are the correct values for an empty data set.
+ val initialCutoffs = Array.fill[Double](quantiles.length)(Double.PositiveInfinity)
+
+ computeEcdf(data).aggregate(initialCutoffs)(addDataPointToKnownCutoffs, mergeCutoffs)
+ }
+
+ /**
+ * Compute the empirical cumulative distribution function,
+ * cf. [[https://en.wikipedia.org/wiki/Empirical_distribution_function]]
+ *
+ * @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 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)
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TimeUtilities.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TimeUtilities.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TimeUtilities.scala
new file mode 100644
index 0000000..1318dbb
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TimeUtilities.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.transformation
+
+object TimeUtilities {
+
+
+ /**
+ * It converts HH:MM:SS string to seconds
+ *
+ * @param timeStr This is time in the form of a string
+ * @return It returns time converted to seconds
+ */
+
+ def getTimeAsDouble(timeStr: String) : Double = {
+ val s = timeStr.split(":")
+ val hours = s(0).toInt
+ val minutes = s(1).toInt
+ val seconds = s(2).toInt
+
+ (3600*hours + 60*minutes + seconds).toDouble
+ }
+
+ /**
+ * It takes only the hour element of time
+ *
+ * @param timeStr This is time in the form of a string
+ * @return It returns only the hour of time
+ */
+ def getTimeAsHour(timeStr: String): Int = {
+ val s = timeStr.split(":")
+ val hours = s(0).toInt
+ hours
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TopDomains.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TopDomains.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TopDomains.scala
new file mode 100644
index 0000000..c0ac20f
--- /dev/null
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/transformation/TopDomains.scala
@@ -0,0 +1,31 @@
+/*
+ * 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.transformation
+
+import scala.io.Source
+
+
+object TopDomains {
+ val alexaTop1MPath = "top-1m.csv"
+
+ val TopDomains: Set[String] = Source.fromFile(alexaTop1MPath).getLines.map(line => {
+ val parts = line.split(",")
+ val l = parts.length
+ parts(1).split('.')(0)
+ }).toSet
+}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala b/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala
deleted file mode 100644
index 54fec85..0000000
--- a/spot-ml/src/test/scala/org/apache/spot/QuantilesTest.scala
+++ /dev/null
@@ -1,178 +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.spark.rdd.RDD
-import org.apache.spot.testutils.TestingSparkContextFlatSpec
-import org.apache.spot.utilities.Quantiles
-import org.scalatest.Matchers
-
-
-class QuantilesTest extends TestingSparkContextFlatSpec with Matchers {
-
- val allOnes = List(1.0, 1.0, 1.0, 1.0, 1.0)
- val onesAndTwos = List(1.0, 2.0, 1.0, 2.0)
- val countToTen = List(1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0)
-
-
- "ecdf" should "work on an empty list" in {
- val rddIn: RDD[Double] = sparkContext.parallelize(List(), numSlices = 3)
- val rddOut = Quantiles.computeEcdf(rddIn)
-
- val out = rddOut.collect()
-
- out.length shouldBe 0
- }
-
- "ecdf" should "work on a constant list" in {
- val rddIn = sparkContext.parallelize(allOnes, numSlices = 3)
- val rddOut = Quantiles.computeEcdf(rddIn)
-
- val out = rddOut.collect()
-
- out.length shouldBe 1
- out(0) shouldBe(1.0, 1.0)
- }
-
- "ecdf" should "work on a split 50/50 list" in {
- val rddIn = sparkContext.parallelize(onesAndTwos, numSlices = 3)
- val rddOut = Quantiles.computeEcdf(rddIn)
-
- val out = rddOut.collect()
-
- out.length shouldBe 2
- out(0) shouldBe(1.0, 0.5)
- out(1) shouldBe(2.0, 1.0)
- }
-
- "ecdf" should "work on count-to-ten list" in {
- val rddIn = sparkContext.parallelize(countToTen, numSlices = 3)
- val rddOut = Quantiles.computeEcdf(rddIn)
-
- val out = rddOut.collect()
-
- out.length shouldBe 10
- out(0) shouldBe(1.0, 0.1)
- out(1) shouldBe(2.0, 0.2)
- out(2) shouldBe(3.0, 0.3)
- out(3) shouldBe(4.0, 0.4)
- out(4) shouldBe(5.0, 0.5)
- out(5) shouldBe(6.0, 0.6)
- out(6) shouldBe(7.0, 0.7)
- out(7) shouldBe(8.0, 0.8)
- out(8) shouldBe(9.0, 0.9)
- out(9) shouldBe(10.0, 1.0)
- }
-
- "quantiles" should "work on an empty list" in {
- val rddIn: RDD[Double] = sparkContext.parallelize(List(), numSlices = 3)
- val quantiles = Array(0.0, 0.5)
- val out = Quantiles.computeQuantiles(rddIn, quantiles)
-
- out.length shouldBe 2
- out(0) shouldBe Double.PositiveInfinity
- out(1) shouldBe Double.PositiveInfinity
- }
-
- "quantiles" should "work on all ones list" in {
- val rddIn = sparkContext.parallelize(allOnes, numSlices = 3)
- val quantiles = Array(0.0, 0.5)
- val out = Quantiles.computeQuantiles(rddIn, quantiles)
-
- out.length shouldBe 2
- out(0) shouldBe 1.0
- out(1) shouldBe 1.0
- }
-
- "quantiles" should "work on a 50/50 1s and 2s list" in {
- val rddIn = sparkContext.parallelize(onesAndTwos, numSlices = 3)
- val quantiles = Array(0.0, 0.5)
- val out = Quantiles.computeQuantiles(rddIn, quantiles)
-
- out.length shouldBe 2
- out(0) shouldBe 1.0
- out(1) shouldBe 1.0
- }
-
- "quantiles" should "work on a 50/50 1s and 2s list with unbalanced thresholds" in {
- val rddIn = sparkContext.parallelize(onesAndTwos, numSlices = 3)
- val quantiles = Array(0.0, 0.6, 0.99)
- val out = Quantiles.computeQuantiles(rddIn, quantiles)
-
- out.length shouldBe 3
- out(0) shouldBe 1.0
- out(1) shouldBe 2.0
- out(2) shouldBe 2.0
- }
-
- "quantiles" should "work on a count-to-ten list" in {
- val rddIn = sparkContext.parallelize(countToTen, numSlices = 3)
- val quantiles = Array(0.0, 0.5)
- val out = Quantiles.computeQuantiles(rddIn, quantiles)
-
- out.length shouldBe 2
- out(0) shouldBe 1.0
- out(1) shouldBe 5.0
-
- }
-
- "deciles" should "work on a count-to-ten list" in {
- val rddIn = sparkContext.parallelize(countToTen, numSlices = 3)
- val out = Quantiles.computeDeciles(rddIn)
-
- out.length shouldBe 10
- out(0) shouldBe 1.0
- out(1) shouldBe 2.0
- out(2) shouldBe 3.0
- out(3) shouldBe 4.0
- out(4) shouldBe 5.0
- out(5) shouldBe 6.0
- out(6) shouldBe 7.0
- out(7) shouldBe 8.0
- out(8) shouldBe 9.0
- out(9) shouldBe 10.0
- }
-
- "quintiles" should "work on a count-to-ten list" in {
- val rddIn = sparkContext.parallelize(countToTen, numSlices = 3)
- val out = Quantiles.computeQuintiles(rddIn)
-
- out.length shouldBe 5
- out(0) shouldBe 2.0
- out(1) shouldBe 4.0
- out(2) shouldBe 6.0
- out(3) shouldBe 8.0
- out(4) shouldBe 10.0
- }
-
- "bin" should "return 3 when the value is not bigger than the fourth quintile" in {
- val quintiles = Array(1.0, 2.0, 3.0, 4.0, 5.0)
-
- val result = Quantiles.bin(3.5, quintiles)
-
- result shouldBe 3
- }
-
- it should "return 0 when the value is less than the first quintile" in {
- val quintiles = Array(1.0, 2.0, 3.0, 4.0, 5.0)
-
- val result = Quantiles.bin(0.0, quintiles)
-
- result shouldBe 0
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/SpotLDAWrapperTest.scala
----------------------------------------------------------------------
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 8baaf34..0000000
--- 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
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/SuspiciousConnectsScoreFunctionTest.scala
----------------------------------------------------------------------
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 0000000..1412716
--- /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.transformation.{ProbabilityConverterDouble, ProbabilityConverterFloat}
+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 probabilityConversionOption = ProbabilityConverterDouble
+ val documentProbabilities = Seq(0.05d, 0.05d, 0.05d, 0.05d)
+
+ val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC)
+
+ val score = scoreFunction.score(probabilityConversionOption)(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 probabilityConversionOption = ProbabilityConverterFloat
+ val documentProbabilities = Seq(0.05f, 0.05f, 0.05f, 0.05f)
+
+ val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC)
+
+ val score = scoreFunction.score(probabilityConversionOption)(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 probabilityConversionOption = ProbabilityConverterFloat
+ val documentProbabilities = Seq(0.05f, 0.05f, 0.05f, 0.05f)
+
+ val scoreFunction = new SuspiciousConnectsScoreFunction(topicCount, wordToPerTopicProbBC)
+
+ val score = scoreFunction.score(probabilityConversionOption)(documentProbabilities, "word_error")
+
+ score shouldBe -1d
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
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 699f954..681fe8c 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
@@ -25,6 +25,7 @@ import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.dns.DNSSchema._
import org.apache.spot.dns.model.DNSSuspiciousConnectsModel
import org.apache.spot.testutils.TestingSparkContextFlatSpec
+import org.apache.spot.utilities.transformation.ProbabilityConverterFloat
import org.scalatest.Matchers
@@ -53,6 +54,21 @@ 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,
+ probabilityConversionOption = ProbabilityConverterFloat)
+
"dns suspicious connects analysis" should "estimate correct probabilities in toy data with framelength anomaly" in {
@@ -76,7 +92,7 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with
}
- "dns suspicious 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)
@@ -114,6 +130,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
+ }
"filterRecords" should "return data set without garbage" in {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala b/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
index b9e0a2d..52710d0 100644
--- a/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
@@ -18,7 +18,7 @@
package org.apache.spot.dns
import org.apache.spot.testutils.TestingSparkContextFlatSpec
-import org.apache.spot.utilities.Entropy
+import org.apache.spot.utilities.transformation.Entropy
import org.scalatest.Matchers
class DNSWordCreationTest extends TestingSparkContextFlatSpec with Matchers {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/lda/SpotLDAWrapperTest.scala
----------------------------------------------------------------------
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 0000000..2660ef7
--- /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.transformation.{ProbabilityConverterDouble, ProbabilityConverterFloat}
+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, ProbabilityConverterDouble)
+
+ 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, ProbabilityConverterFloat)
+
+ 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, ProbabilityConverterDouble)
+
+ 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, ProbabilityConverterFloat)
+
+ 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 ProbabilityConverterDouble: 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,
+ ProbabilityConverterDouble)
+
+ 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 ProbabilityConverterFloat: 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,
+ ProbabilityConverterFloat)
+
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index 4da54c4..4897e3a 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
@@ -25,6 +25,7 @@ import org.apache.spot.netflow.FlowSchema._
import org.apache.spot.netflow.FlowSuspiciousConnectsAnalysis.InSchema
import org.apache.spot.netflow.model.FlowSuspiciousConnectsModel
import org.apache.spot.testutils.TestingSparkContextFlatSpec
+import org.apache.spot.utilities.transformation.ProbabilityConverterFloat
import org.scalatest.Matchers
@@ -46,6 +47,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,
+ probabilityConversionOption = ProbabilityConverterFloat)
+
"netflow suspicious connects" should "correctly identify time-of-day anomalies" in {
val logger = LogManager.getLogger("SuspiciousConnectsAnalysis")
@@ -82,7 +98,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
- "netflow suspicious connects" should "correctly identify time-of-day anomalies with testing config" in {
+ it should "correctly identify time-of-day anomalies with testing config" in {
val testConfig2 = SuspiciousConnectsConfig(analysis = "flow",
inputPath = "",
@@ -136,6 +152,45 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
}
+ 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))
+
+ 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
+
+
+ }
+
"filterRecords" should "return data set without garbage" in {
val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
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 04d5a46..d3f90eb 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
@@ -23,32 +23,32 @@ 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.transformation.ProbabilityConverterFloat
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,28 @@ 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,
+ probabilityConversionOption = ProbabilityConverterFloat)
"proxy supicious 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))
@@ -94,19 +96,52 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi
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
+ }
"filterRecords" should "return data without garbage" in {
@@ -143,18 +178,18 @@ class ProxySuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wi
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(
@@ -181,16 +216,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(
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.scala
----------------------------------------------------------------------
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
deleted file mode 100644
index 0dadfc0..0000000
--- a/spot-ml/src/test/scala/org/apache/spot/utilities/DomainProcessorTest.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.utilities
-
-import org.apache.spot.testutils.TestingSparkContextFlatSpec
-import org.scalatest.{FunSuite, Matchers}
-import org.apache.spot.utilities.DomainProcessor._
-
-
-class DomainProcessorTest extends TestingSparkContextFlatSpec with Matchers {
- val countryCodesSet = CountryCodes.CountryCodes
-
-
- "extractDomain" should "return domain when provided a url with top-level domain and country code" in {
- val url = "fatosdesconhecidos.com.br"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("fatosdesconhecidos")
- }
-
- it should "return domain when provided a short url with no top-level domain but only a country code" in {
- val url = "panasonic.jp"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("panasonic")
- }
-
- it should "return domain when provided a long url with no top-level domain but only a country code" in {
- val url = "get.your.best.electronic.at.panasonic.jp"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("panasonic")
- }
-
- it should "return domain when provided a short url with a top-level domain no country code" in {
- val url = "forrealz.net"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("forrealz")
- }
-
- it should "return domain when provided a long url with a top-level domain no country code" in {
- val url = "wow.its.really.long.super.long.yeah.so.long.long.long.long.forrealz.net"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("forrealz")
- }
-
- it should "should return \"None\" when provided an address" in {
- val url = "123.103.104.10.in-addr.arpa"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("None")
- }
-
- it should "return \"None\" when provided a short url with a bad top-level domain / country code" in {
- val url = "panasonic.c"
- val result = DomainProcessor.extractDomain(url)
- result shouldEqual ("None")
- }
-
- "extractDomainInfo" should "handle an in-addr.arpa url" in {
-
- val url = "123.103.104.10.in-addr.arpa"
-
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
-
- val userDomain = "intel"
-
- // case class DerivedFields(topDomain: String, subdomainLength: Double, subdomainEntropy: Double, numPeriods: Double)
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "None", topDomain = 0, subdomain = "None", subdomainLength = 0, subdomainEntropy = 0, numPeriods = 6)
- }
-
- it should "handle an Alexa top 1M domain with a subdomain, top-level domain name and country code" in {
-
- val url = "services.amazon.com.mx"
-
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
-
- val userDomain = "intel"
-
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "amazon", topDomain = 1, subdomain = "services",
- subdomainLength = 8, subdomainEntropy = 2.5, numPeriods = 4)
- }
-
- it should "handle an Alexa top 1M domain with a top-level domain name and country code but no subdomain" in {
-
- val url = "amazon.com.mx"
- val countryCodes = sparkContext.broadcast(countryCodesSet)
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
- val userDomain = "intel"
-
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "amazon", subdomain = "None", topDomain = 1, subdomainLength = 0, subdomainEntropy = 0, numPeriods = 3)
- }
-
- it should "handle an Alexa top 1M domain with a subdomain and top-level domain name but no country code" in {
-
- val url = "services.amazon.com"
- val countryCodes = sparkContext.broadcast(countryCodesSet)
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
- val userDomain = "intel"
-
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "amazon", subdomain = "services", topDomain = 1, subdomainLength = 8, subdomainEntropy = 2.5, numPeriods = 3)
- }
-
-
- it should "handle an Alexa top 1M domain with no subdomain or country code" in {
-
- val url = "amazon.com"
- val countryCodes = sparkContext.broadcast(countryCodesSet)
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
- val userDomain = "intel"
-
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "amazon", subdomain = "None", topDomain = 1, subdomainLength = 0, subdomainEntropy = 0, numPeriods = 2)
- }
- it should "not identify the domain as the users domain when both are empty strings" in {
- val url = "ab..com"
- val countryCodes = sparkContext.broadcast(countryCodesSet)
- val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
- val userDomain = ""
-
- val result = extractDomainInfo(url, topDomains, userDomain)
-
- result shouldBe DomainInfo(domain = "", subdomain = "ab", topDomain = 0, subdomainLength = 2, subdomainEntropy = 1, numPeriods = 3)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/DomainProcessorTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/DomainProcessorTest.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/DomainProcessorTest.scala
new file mode 100644
index 0000000..0bd50ac
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/DomainProcessorTest.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.transformation
+
+import org.apache.spot.testutils.TestingSparkContextFlatSpec
+import org.apache.spot.utilities.transformation
+import org.apache.spot.utilities.transformation.DomainProcessor._
+import org.scalatest.Matchers
+
+
+class DomainProcessorTest extends TestingSparkContextFlatSpec with Matchers {
+ val countryCodesSet = transformation.CountryCodes.CountryCodes
+
+
+ "extractDomain" should "return domain when provided a url with top-level domain and country code" in {
+ val url = "fatosdesconhecidos.com.br"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("fatosdesconhecidos")
+ }
+
+ it should "return domain when provided a short url with no top-level domain but only a country code" in {
+ val url = "panasonic.jp"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("panasonic")
+ }
+
+ it should "return domain when provided a long url with no top-level domain but only a country code" in {
+ val url = "get.your.best.electronic.at.panasonic.jp"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("panasonic")
+ }
+
+ it should "return domain when provided a short url with a top-level domain no country code" in {
+ val url = "forrealz.net"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("forrealz")
+ }
+
+ it should "return domain when provided a long url with a top-level domain no country code" in {
+ val url = "wow.its.really.long.super.long.yeah.so.long.long.long.long.forrealz.net"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("forrealz")
+ }
+
+ it should "should return \"None\" when provided an address" in {
+ val url = "123.103.104.10.in-addr.arpa"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("None")
+ }
+
+ it should "return \"None\" when provided a short url with a bad top-level domain / country code" in {
+ val url = "panasonic.c"
+ val result = DomainProcessor.extractDomain(url)
+ result shouldEqual ("None")
+ }
+
+ "extractDomainInfo" should "handle an in-addr.arpa url" in {
+
+ val url = "123.103.104.10.in-addr.arpa"
+
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+
+ val userDomain = "intel"
+
+ // case class DerivedFields(topDomain: String, subdomainLength: Double, subdomainEntropy: Double, numPeriods: Double)
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "None", topDomain = 0, subdomain = "None", subdomainLength = 0, subdomainEntropy = 0, numPeriods = 6)
+ }
+
+ it should "handle an Alexa top 1M domain with a subdomain, top-level domain name and country code" in {
+
+ val url = "services.amazon.com.mx"
+
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+
+ val userDomain = "intel"
+
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "amazon", topDomain = 1, subdomain = "services",
+ subdomainLength = 8, subdomainEntropy = 2.5, numPeriods = 4)
+ }
+
+ it should "handle an Alexa top 1M domain with a top-level domain name and country code but no subdomain" in {
+
+ val url = "amazon.com.mx"
+ val countryCodes = sparkContext.broadcast(countryCodesSet)
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+ val userDomain = "intel"
+
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "amazon", subdomain = "None", topDomain = 1, subdomainLength = 0, subdomainEntropy = 0, numPeriods = 3)
+ }
+
+ it should "handle an Alexa top 1M domain with a subdomain and top-level domain name but no country code" in {
+
+ val url = "services.amazon.com"
+ val countryCodes = sparkContext.broadcast(countryCodesSet)
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+ val userDomain = "intel"
+
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "amazon", subdomain = "services", topDomain = 1, subdomainLength = 8, subdomainEntropy = 2.5, numPeriods = 3)
+ }
+
+
+ it should "handle an Alexa top 1M domain with no subdomain or country code" in {
+
+ val url = "amazon.com"
+ val countryCodes = sparkContext.broadcast(countryCodesSet)
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+ val userDomain = "intel"
+
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "amazon", subdomain = "None", topDomain = 1, subdomainLength = 0, subdomainEntropy = 0, numPeriods = 2)
+ }
+ it should "not identify the domain as the users domain when both are empty strings" in {
+ val url = "ab..com"
+ val countryCodes = sparkContext.broadcast(countryCodesSet)
+ val topDomains = sparkContext.broadcast(TopDomains.TopDomains)
+ val userDomain = ""
+
+ val result = extractDomainInfo(url, topDomains, userDomain)
+
+ result shouldBe DomainInfo(domain = "", subdomain = "ab", topDomain = 0, subdomainLength = 2, subdomainEntropy = 1, numPeriods = 3)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterDoubleTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterDoubleTest.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterDoubleTest.scala
new file mode 100644
index 0000000..50e46b4
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterDoubleTest.scala
@@ -0,0 +1,48 @@
+package org.apache.spot.utilities.transformation
+
+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 ProbabilityConverterDoubleTest extends TestingSparkContextFlatSpec with Matchers {
+
+ "convertProbability" should "just return the same value with the same type" in {
+ val testValue: Double = 5d
+
+ val result = ProbabilityConverterDouble.convertProbability(testValue)
+
+ result shouldBe testValue
+ result shouldBe a[java.lang.Double]
+
+ }
+
+ "convertBackSetOfProbabilities" should "return a array of the same type" in {
+
+ val testSeq: Seq[Double] = Seq(1d, 2d, 3d)
+
+ val result = ProbabilityConverterDouble.convertBackSetOfProbabilities(testSeq)
+
+ result shouldBe a[Array[Double]]
+ result.length shouldBe 3
+ }
+
+ "convertDataFrameColumn" 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 = ProbabilityConverterDouble.convertDataFrameColumn(testDataFrame, TopicProbabilityMix)
+
+ val schema = StructType(
+ Array(StructField(DocumentName, StringType, true),
+ StructField(TopicProbabilityMix, ArrayType(DoubleType, false), true)))
+
+ result shouldBe testDataFrame
+ result.schema shouldBe schema
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/89884430/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterFloatTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterFloatTest.scala b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterFloatTest.scala
new file mode 100644
index 0000000..6aa949c
--- /dev/null
+++ b/spot-ml/src/test/scala/org/apache/spot/utilities/transformation/ProbabilityConverterFloatTest.scala
@@ -0,0 +1,49 @@
+package org.apache.spot.utilities.transformation
+
+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 ProbabilityConverterFloatTest extends TestingSparkContextFlatSpec with Matchers {
+
+ "convertProbability" should "just return value converted to Float" in {
+ val testValue: Double = 5d
+
+ val result = ProbabilityConverterFloat.convertProbability(testValue)
+
+ result shouldBe 5f
+ result shouldBe a[java.lang.Float]
+
+ }
+
+ "convertBackSetOfProbabilities" should "convert a Seq of Float to Seq of Double" in {
+
+ val testSeq: Seq[Float] = Seq(1f, 2f, 3f)
+
+ val result = ProbabilityConverterFloat.convertBackSetOfProbabilities(testSeq)
+
+ result(0) shouldBe a[java.lang.Double]
+ result(1) shouldBe a[java.lang.Double]
+ result(2) shouldBe a[java.lang.Double]
+ }
+
+ "convertDataFrameColumn" 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 = ProbabilityConverterFloat.convertDataFrameColumn(testDataFrame, TopicProbabilityMix)
+ result.count
+
+ val schema = StructType(
+ Array(StructField(DocumentName, StringType, true),
+ StructField(TopicProbabilityMix, ArrayType(FloatType, false), true)))
+
+ result.schema shouldBe schema
+
+ }
+}