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_r399740772
##########
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:
```degreesOfFreedom = numSamples - 1```, right? no need to call
```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:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
