大家好，我有以下在Spark 1.6中可以工作的代码。
import org.apache.spark.ml.feature.{ChiSqSelectorModel,QuantileDiscretizer,VectorAssembler,ChiSqSelector}
import org.apache.spark.sql.types.{StructType,StructField,IntegerType}
import org.apache.spark.storage.StorageLevel._
import org.apache.spark.ml.Pipeline
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import scala.util.Random
import scala.math
val nRows = 1000
val nCols = 100
val rD = sc.parallelize(0 to nRows-1,172).map { _ => Row.fromSeq(Seq.fill(nCols)(Random.nextInt(10))) }
val schema = StructType((0 to nCols-1).map { i => StructField("C" + i, IntegerType, true) } )
val df = spark.createDataFrame(rD, schema)
val continuous = df.drop("C0").dtypes.filter (_._2 != "StringType") map (_._1)
val discretizers = continuous .map(c => new QuantileDiscretizer().setInputCol(c).setOutputCol(s"${c}_disc").setNumBuckets(10))
val conDisc = continuous.map(c => s"${c}_disc")
val assembler = new VectorAssembler().setInputCols(conDisc).setOutputCol("features")
val selector = new ChiSqSelector().setNumTopFeatures(100).setFeaturesCol("features").setLabelCol("C0").setOutputCol("selectedFeatures")
val pipeline = new Pipeline().setStages(Array.concat(discretizers.toArray, Array(assembler, selector)))
val model = pipeline.fit(df)如何将其转换为在2.0版本中工作？问题似乎出在discretizers上。Spark抛出以下错误
<console>:56: error: type mismatch; found : Array[org.apache.spark.ml.feature.QuantileDiscretizer] required: Array[org.apache.spark.ml.PipelineStage with org.apache.spark.ml.para m.shared.HasOutputCol with org.apache.spark.ml.util.DefaultParamsWritable{def co py(extra: org.apache.spark.ml.param.ParamMap): org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.uti l.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.ParamMap): org .apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.HasOutputCo l with org.apache.spark.ml.util.DefaultParamsWritable}}] Note: org.apache.spark.ml.feature.QuantileDiscretizer <: org.apache.spark.ml.Pip elineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.sp ark.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.Para mMap): org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.H asOutputCol with org.apache.spark.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.ParamMap): org.apache.spark.ml.PipelineStage with org. apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.util.DefaultP aramsWritable}}, but class Array is invariant in type T. You may wish to investigate a wildcard type such as `_ <: org.apache.spark.ml.Pi pelineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.s park.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.Par amMap): org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared. HasOutputCol with org.apache.spark.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.ParamMap): org.apache.spark.ml.PipelineStage with org .apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.util.Default ParamsWritable}}`. (SLS 3.2.10) val pipeline = new Pipeline().setStages(Array.concat(discretizers.toArray , Array(assembler, selector)))感谢帮助。
回答：
希望这对你有帮助：
val pipeline = new Pipeline().setStages(discretizers ++ Array(assembler, selector))