package ml import java.util import org.apache.spark.mllib.linalg.{Vector, Vectors} import org.apache.spark.mllib.linalg.distributed.RowMatrix import org.apache.spark.rdd.RDD import org.apache.spark.sql.{DataFrame, SQLContext, Row} import org.apache.spark.{SparkContext, SparkConf} import java.util.Arrays import org.apache.spark.ml.attribute.{Attribute, AttributeGroup, NumericAttribute} import org.apache.spark.ml.feature.VectorSlicer import org.apache.spark.sql.types.{DataTypes, StructField, StructType} /* VectorSlicer是一个转换器,输入一个特征向量输出一个特征向量, 它是原特征的一个子集。这在从向量列中抽取特征非常有用。 VectorSlicer接收一个拥有特定索引的特征列, 它的输出是一个新的特征列,它的值通过输入的索引来选择。 有两种类型的索引: 1、整数索引表示进入向量的索引,调用setIndices() 2、字符串索引表示进入向量的特征列的名称,调用setNames()。 这种情况需要向量列拥有一个AttributeGroup,这是因为实现是通过属性的名字来匹配的。 * */ object FeatureSelectors {
      def main(args: Array[String]) {
          val conf = new SparkConf().setAppName("test").setMaster("local")         val sc = new SparkContext(conf)         val sql = new SQLContext(sc);         val data = Arrays.asList( //            Row(Vectors.dense(-2.0, 2, 0.0)),             Row(Vectors.sparse(3, Seq((0, -2.0), (1, 2.3)))),             Row(Vectors.dense(-2.0, 2, 0.0))         )         val defaultAttr: NumericAttribute = NumericAttribute.defaultAttr         val attrs = Array("f1", "f2", "f3").map(defaultAttr.withName)         val attrGroup = new AttributeGroup("userFeatures", attrs.asInstanceOf[Array[Attribute]]) //从三列中选择两列参与模型训练         val dataset = sql.createDataFrame(data, StructType(Array(attrGroup.toStructField())))         dataset.printSchema()         val slicer = new VectorSlicer().setInputCol("userFeatures").setOutputCol("features")         //setIndices(Array(1)) 第二列   setNames(Array("f3")) 第三列 //        slicer.setIndices(Array(1)).setNames(Array("f3"))         slicer.setIndices(Array(1)).setNames(Array("f3"))         // or slicer.setIndices(Array(1, 2)), or slicer.setNames(Array("f2", "f3"))         val output: DataFrame = slicer.transform(dataset)         output.printSchema()         output.show(false)         output.select("features").show() //        val out: RDD[Row] = output.rdd.map(row => Row(row.get(0),row.get(1)))         val out: DataFrame = output.select("features")         val rdd: RDD[Row] = out.toDF().map{ row =>             val r: Vector = row.getAs[Vector](0)         Row(r.apply(0),r.apply(1)) //        println("---"+r.apply(0)+"---"+r.apply(1))         }         val fields = new util.ArrayList[StructField];         fields.add(DataTypes.createStructField("id", DataTypes.DoubleType, true));         fields.add(DataTypes.createStructField("feature", DataTypes.DoubleType, true));         val structType = DataTypes.createStructType(fields);         sql.createDataFrame(rdd,structType).show()     } }