Spark + scala new pipline for StringIndexer multiple columns

Question

我尝试在多个列上应用StringIndexer()，我使用Scala和Spark 2.3。
这是我的代码：

val df1 = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("file:///c:/tmp/spark-warehouse/train.csv")
val feat = df1.columns.filterNot(_ .contains("BsmtFinSF1"))
val inds = feat.map { colName =>
  val indexer1 = new StringIndexer()
    .setInputCol(colName)
    .setOutputCol(colName + "I")
    .fit(df1)
  Array(indexer1)
}
val pipeline = new Pipeline().setStages(inds.toArray)

但是，我有这个错误：

错误：（134， 50） 类型不匹配;

找到 ： Array[Array[org.apache.spark.ml.feature.StringIndexerModel]]
必需：数组[？ <： org.apache.spark.ml.PipelineStage]

注意：Array[org.apache.spark.ml.feature.StringIndexerModel]>： ？ <：org.apache.spark.ml.PipelineStage，但类 Array 在类型 T 中是不变的。 您可能希望调查通配符类型，例如 _ >: ? <: org.apache.spark.ml.PipelineStage 。（SLS 3.2.10）
val pipeline = new Pipeline（）.setStages（inds.toArray）

任何帮助将不胜感激。谢谢

Answer 1

.setStages需要Array[PipelineStage]，但实际上它变得Array[Array[PipelineStage]，因为您将indexer1包装到冗余数组中：Array(indexer1)。映射函数返回相同类型的集合。此集合的元素是由传递给 Map 的函数的应用产生的。所以试试这样：

val inds = feat.map { colName =>
   new StringIndexer()
    .setInputCol(colName)
    .setOutputCol(colName + "I")
    .fit(df1)          
}