Spark + scala новый конвейер для множителей StringIndexer

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]]
обязательно: Array [?<: org.apache.spark.ml.PipelineStage] </p>

Примечание. Массив [org.apache.spark.ml.feature.StringIndexerModel]>:?<: org.apache.spark.ml.PipelineStage, но класс Array инвариантен в типе T. Возможно, вы захотите исследовать подстановочный тип, такой как <code>_ >: ? <: org.apache.spark.ml.PipelineStage.(SLS 3.2.10)
val pipe = новый Pipeline (). SetStages (inds.toArray)

Любая помощь будет оценена.спасибо

Answer 1

.setStages занимает Array[PipelineStage], но на самом деле оно становится Array[Array[PipelineStage], потому что вы помещаете indexer1 в избыточный массив здесь: Array(indexer1).Функция Map возвращает коллекцию того же типа.Элементы этой коллекции создаются приложением функции, переданной в Map.Просто попробуйте вот так:

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