所有， 我有以下适用于 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 中工作？ 问题似乎出在离散器上。 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))
关于scala - 在 1.6 中工作的 Spark ml 管道在 2.0 中不起作用。类型不匹配错误，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43009257/