[SPARK-16843][MLLIB] add the percentage ChiSquareSelector feature

mllib/src/main/scala/org/apache/spark/mllib/feature/ChiSqSelector.scala

@@ -197,3 +197,31 @@ class ChiSqSelector @Since("1.3.0") (
     new ChiSqSelectorModel(indices)
   }
 }
+/**
+ * Creates a ChiSquared feature selector by Percentile.
+ * @param percentile percentage of features that selector will select
+ *                   (ordered by statistic value descending)
+ *                   Note that if the percentile is larger than 100,
+ *                   then this will select all features.
+ */
+@Since("2.0.0")
+class PercentileChiSqSelector @Since("2.0.0") (
+  @Since("2.0.0") val percentile: Int) extends Serializable {
+
+  /**
+   * Returns a ChiSquared feature selector.
+   *
+   * @param data an `RDD[LabeledPoint]` containing the labeled dataset with categorical features.
+   *             Real-valued features will be treated as categorical for each distinct value.
+   *             Apply feature discretizer before using this function.
+   */
+  @Since("2.0.0")
+  def fit(data: RDD[LabeledPoint]): ChiSqSelectorModel = {
+    val indices = Statistics.chiSqTest(data)
+      .zipWithIndex.sortBy { case (res, _) => -res.statistic }
+      .take((data.count() * percentile / 100).toInt)
+      .map { case (_, indices) => indices }
+      .sorted
+    new ChiSqSelectorModel(indices)
+  }
+}

mllib/src/test/scala/org/apache/spark/mllib/feature/ChiSqSelectorSuite.scala

@@ -65,6 +65,24 @@ class ChiSqSelectorSuite extends SparkFunSuite with MLlibTestSparkContext {
     assert(filteredData == preFilteredData)
   }
 
+  test("PercentileChiSqSelector transform test (sparse & dense vector)") {
+    val labeledDiscreteData = sc.parallelize(
+      Seq(LabeledPoint(0.0, Vectors.sparse(3, Array((0, 8.0), (1, 7.0)))),
+        LabeledPoint(1.0, Vectors.sparse(3, Array((1, 9.0), (2, 6.0)))),
+        LabeledPoint(1.0, Vectors.dense(Array(0.0, 9.0, 8.0))),
+        LabeledPoint(2.0, Vectors.dense(Array(8.0, 9.0, 5.0)))), 2)
+    val preFilteredData =
+      Set(LabeledPoint(0.0, Vectors.dense(Array(0.0))),
+        LabeledPoint(1.0, Vectors.dense(Array(6.0))),
+        LabeledPoint(1.0, Vectors.dense(Array(8.0))),
+        LabeledPoint(2.0, Vectors.dense(Array(5.0))))
+    val model = new PercentileChiSqSelector(25).fit(labeledDiscreteData)
+    val filteredData = labeledDiscreteData.map { lp =>
+      LabeledPoint(lp.label, model.transform(lp.features))
+    }.collect().toSet
+    assert(filteredData == preFilteredData)
+  }
+
   test("model load / save") {
     val model = ChiSqSelectorSuite.createModel()
     val tempDir = Utils.createTempDir()