You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/03/26 22:07:44 UTC
[GitHub] [spark] huaxingao commented on a change in pull request #27982:
[SPARK-31222][ML] Make ANOVATest Sparsity-Aware
huaxingao commented on a change in pull request #27982: [SPARK-31222][ML] Make ANOVATest Sparsity-Aware
URL: https://github.com/apache/spark/pull/27982#discussion_r398921906
##########
File path: mllib/src/main/scala/org/apache/spark/ml/stat/ANOVATest.scala
##########
@@ -80,65 +81,141 @@ object ANOVATest {
SchemaUtils.checkColumnType(dataset.schema, featuresCol, new VectorUDT)
SchemaUtils.checkNumericType(dataset.schema, labelCol)
- dataset.select(col(labelCol).cast("double"), col(featuresCol))
- .as[(Double, Vector)]
- .rdd
- .flatMap { case (label, features) =>
- features.iterator.map { case (col, value) => (col, (label, value)) }
- }.aggregateByKey[(Double, Double, OpenHashMap[Double, Double], OpenHashMap[Double, Long])](
- (0.0, 0.0, new OpenHashMap[Double, Double], new OpenHashMap[Double, Long]))(
- seqOp = {
- case ((sum, sumOfSq, sums, counts), (label, value)) =>
- // sums: mapOfSumPerClass (key: label, value: sum of features for each label)
- // counts: mapOfCountPerClass key: label, value: count of features for each label
- sums.changeValue(label, value, _ + value)
- counts.changeValue(label, 1L, _ + 1L)
- (sum + value, sumOfSq + value * value, sums, counts)
- },
- combOp = {
- case ((sum1, sumOfSq1, sums1, counts1), (sum2, sumOfSq2, sums2, counts2)) =>
- sums2.foreach { case (v, w) => sums1.changeValue(v, w, _ + w) }
- counts2.foreach { case (v, w) => counts1.changeValue(v, w, _ + w) }
- (sum1 + sum2, sumOfSq1 + sumOfSq2, sums1, counts1)
- }
- ).map { case (col, (sum, sumOfSq, sums, counts)) =>
- val numSamples = counts.iterator.map(_._2).sum
- val numClasses = counts.size
-
- // e.g. features are [3.3, 2.5, 1.0, 3.0, 2.0] and labels are [1, 2, 1, 3, 3]
- // sum: sum of all the features (3.3+2.5+1.0+3.0+2.0)
- // sumOfSq: sum of squares of all the features (3.3^2+2.5^2+1.0^2+3.0^2+2.0^2)
+ val points = dataset.select(col(labelCol).cast("double"), col(featuresCol))
+ .as[(Double, Vector)].rdd
+
+ points.first()._2 match {
+ case dv: DenseVector =>
+ testClassificationDenseFeatures(points, dv.size)
+ case sv: SparseVector =>
+ testClassificationSparseFeatures(points, sv.size)
+ }
+ }
+
+ private def testClassificationDenseFeatures(
+ points: RDD[(Double, Vector)],
+ numFeatures: Int): Array[SelectionTestResult] = {
+ points.flatMap { case (label, features) =>
+ require(features.size == numFeatures,
+ s"Number of features must be $numFeatures but got ${features.size}")
+ features.iterator.map { case (col, value) => (col, (label, value)) }
+ }.aggregateByKey[(Double, Double, OpenHashMap[Double, Double], OpenHashMap[Double, Long])](
+ (0.0, 0.0, new OpenHashMap[Double, Double], new OpenHashMap[Double, Long]))(
+ seqOp = {
+ case ((sum, sumOfSq, sums, counts), (label, value)) =>
+ // sums: mapOfSumPerClass (key: label, value: sum of features for each label)
+ // counts: mapOfCountPerClass key: label, value: count of features for each label
+ sums.changeValue(label, value, _ + value)
+ counts.changeValue(label, 1L, _ + 1L)
+ (sum + value, sumOfSq + value * value, sums, counts)
+ },
+ combOp = {
+ case ((sum1, sumOfSq1, sums1, counts1), (sum2, sumOfSq2, sums2, counts2)) =>
+ sums2.foreach { case (v, w) => sums1.changeValue(v, w, _ + w) }
+ counts2.foreach { case (v, w) => counts1.changeValue(v, w, _ + w) }
+ (sum1 + sum2, sumOfSq1 + sumOfSq2, sums1, counts1)
+ }
+ ).mapValues { case (sum, sumOfSq, sums, counts) =>
+ computeANOVA(sum, sumOfSq, sums.toMap, counts.toMap)
+ }.collect().sortBy(_._1).map {
+ case (_, (pValue, degreesOfFreedom, fValue)) =>
+ new ANOVATestResult(pValue, degreesOfFreedom, fValue)
+ }
+ }
+
+ private def testClassificationSparseFeatures(
+ points: RDD[(Double, Vector)],
+ numFeatures: Int): Array[SelectionTestResult] = {
+ val sc = points.sparkContext
+ val counts = points.map(_._1).countByValue().toMap
+ val bcCounts = sc.broadcast(counts)
+
+ val results = points.flatMap { case (label, features) =>
+ require(features.size == numFeatures,
+ s"Number of features must be $numFeatures but got ${features.size}")
+ features.nonZeroIterator.map { case (col, value) => (col, (label, value)) }
+ }.aggregateByKey[(Double, Double, OpenHashMap[Double, Double])](
+ (0.0, 0.0, new OpenHashMap[Double, Double]))(
+ seqOp = {
+ case ((sum, sumOfSq, sums), (label, value)) =>
// sums: mapOfSumPerClass (key: label, value: sum of features for each label)
- // ( 1 -> 3.3 + 1.0, 2 -> 2.5, 3 -> 3.0 + 2.0 )
- // counts: mapOfCountPerClass (key: label, value: count of features for each label)
- // ( 1 -> 2, 2 -> 2, 3 -> 2 )
- // sqSum: square of sum of all data ((3.3+2.5+1.0+3.0+2.0)^2)
- val sqSum = sum * sum
- val ssTot = sumOfSq - sqSum / numSamples
-
- // sumOfSqSumPerClass:
- // sum( sq_sum_classes[k] / n_samples_per_class[k] for k in range(n_classes))
- // e.g. ((3.3+1.0)^2 / 2 + 2.5^2 / 1 + (3.0+2.0)^2 / 2)
- val sumOfSqSumPerClass = sums.iterator
- .map { case (label, sum) => sum * sum / counts(label) }.sum
- // Sums of Squares Between
- val ssbn = sumOfSqSumPerClass - (sqSum / numSamples)
- // Sums of Squares Within
- val sswn = ssTot - ssbn
- // degrees of freedom between
- val dfbn = numClasses - 1
- // degrees of freedom within
- val dfwn = numSamples - numClasses
- // mean square between
- val msb = ssbn / dfbn
- // mean square within
- val msw = sswn / dfwn
- val fValue = msb / msw
- val pValue = 1 - new FDistribution(dfbn, dfwn).cumulativeProbability(fValue)
- (col, pValue, dfbn + dfwn, fValue)
- }.collect().sortBy(_._1).map {
- case (col, pValue, degreesOfFreedom, fValue) =>
- new ANOVATestResult(pValue, degreesOfFreedom, fValue)
- }
+ sums.changeValue(label, value, _ + value)
+ (sum + value, sumOfSq + value * value, sums)
+ },
+ combOp = {
+ case ((sum1, sumOfSq1, sums1), (sum2, sumOfSq2, sums2)) =>
+ sums2.foreach { case (v, w) => sums1.changeValue(v, w, _ + w) }
+ (sum1 + sum2, sumOfSq1 + sumOfSq2, sums1)
+ }
+ ).mapValues { case (sum, sumOfSq, sums) =>
+ val counts = bcCounts.value
+ counts.keysIterator.foreach { label =>
+ // adjust sums if all related feature values are 0 for some label
+ if (!sums.contains(label)) sums.update(label, 0.0)
+ }
+ computeANOVA(sum, sumOfSq, sums.toMap, counts)
+ }.collectAsMap()
+
+ bcCounts.destroy()
+
+ val finalResults = Array.ofDim[SelectionTestResult](numFeatures)
+ results.foreach { case (col, (pValue, degreesOfFreedom, fValue)) =>
+ finalResults(col) = new ANOVATestResult(pValue, degreesOfFreedom, fValue)
+ }
+
+ if (results.size < numFeatures) {
+ // if some column only contains 0 values
+ val (pValue, degreesOfFreedom, fValue) =
+ computeANOVA(0.0, 0.0, counts.mapValues(_ => 0.0), counts)
Review comment:
For the category that only contains value 0s, the ```pValue``` and ```fValue``` are NaN. Maybe just get the DF and skip calling ```computeANOVA```?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org