Stosowanie indeksu IndexToString do wektora funkcji w Spark

Question

Kontekst: Mam ramkę danych, w której wszystkie wartości kategoryczne zostały zindeksowane przy użyciu StringIndexer.

val categoricalColumns = df.schema.collect { case StructField(name, StringType, nullable, meta) => name }  

val categoryIndexers = categoricalColumns.map { 
    col => new StringIndexer().setInputCol(col).setOutputCol(s"${col}Indexed") 
} 

Następnie użyłem VectorAssembler do wektoryzacji wszystkich kolumn elementów (w tym indeksowanych kategorii).

val assembler = new VectorAssembler() 
    .setInputCols(dfIndexed.columns.diff(List("label") ++ categoricalColumns)) 
    .setOutputCol("features") 

Po zastosowaniu klasyfikatora i kilku dodatkowych czynności uzyskuję ramkę danych z etykietą, funkcjami i podpowiedziami. Chciałbym rozwinąć mój wektor funkcji, aby oddzielić kolumny, aby przekonwertować indeksowane wartości z powrotem na ich pierwotną formę String.

val categoryConverters = categoricalColumns.zip(categoryIndexers).map { 
colAndIndexer => new IndexToString().setInputCol(s"${colAndIndexer._1}Indexed").setOutputCol(colAndIndexer._1).setLabels(colAndIndexer._2.fit(df).labels) 
} 

Pytanie: Czy istnieje prosty sposobem osiągnięcia tego, czy jest to najlepsze podejście do jakoś podłączyć kolumny do predykcji ramki danych testowej?

Co próbowałem:

val featureSlicers = categoricalColumns.map { 
    col => new VectorSlicer().setInputCol("features").setOutputCol(s"${col}Indexed").setNames(Array(s"${col}Indexed")) 
} 

Stosując ten daje mi kolumny, które chcę, ale są one w postaci wektorowej (jak to ma zrobić), a nie typu double.

Edycja: żądana jest oryginalny ramki danych (tj kategoryczne cechy jak STRING indeksu) z dodatkową kolumną co wskazuje przewidywany etykietę (moim przypadku jest 0 lub 1).

Na przykład, powiedzmy wyjście mojego klasyfikatora wyglądał mniej więcej tak:

+-----+---------+----------+ 
|label| features|prediction| 
+-----+---------+----------+ 
| 1.0|[0.0,3.0]|  1.0| 
+-----+---------+----------+ 

Stosując VectorSlicer na każdej funkcji chciałbym uzyskać:

+-----+---------+----------+-------------+-------------+ 
|label| features|prediction|statusIndexed|artistIndexed| 
+-----+---------+----------+-------------+-------------+ 
| 1.0|[0.0,3.0]|  1.0|  [0.0]|  [3.0]| 
+-----+---------+----------+-------------+-------------+ 

co jest dobre, ale muszę:

+-----+---------+----------+-------------+-------------+ 
|label| features|prediction|statusIndexed|artistIndexed| 
+-----+---------+----------+-------------+-------------+ 
| 1.0|[0.0,3.0]|  1.0|   0.0 |   3.0 | 
+-----+---------+----------+-------------+-------------+ 

aby następnie móc korzystać IndexToString i przekonwertować go do:

+-----+---------+----------+-------------+-------------+ 
|label| features|prediction| status | artist | 
+-----+---------+----------+-------------+-------------+ 
| 1.0|[0.0,3.0]|  1.0|  good | Pink Floyd | 
+-----+---------+----------+-------------+-------------+ 

lub nawet:

+-----+----------+-------------+-------------+ 
|label|prediction| status | artist | 
+-----+----------+-------------+-------------+ 
| 1.0|  1.0|  good | Pink Floyd | 
+-----+----------+-------------+-------------+ 

Answer 1

Cóż, nie jest to operacja bardzo przydatna, ale powinno być możliwe, aby wyodrębnić potrzebne informacje za pomocą metadanych kolumny i tak proste, UDF.I zakładamy, że dane zostały utworzone rurociągu podobny do tego:

import org.apache.spark.ml.feature.{VectorSlicer, VectorAssembler, StringIndexer} 
import org.apache.spark.ml.Pipeline 

val df = sc.parallelize(Seq(
    (1L, "a", "foo", 1.0), (2L, "b", "bar", 2.0), (3L, "a", "bar", 3.0) 
)).toDF("id", "x1", "x2", "x3") 

val featureCols = Array("x1", "x2", "x3") 
val featureColsIdx = featureCols.map(c => s"${c}_i") 

val indexers = featureCols.map(
    c => new StringIndexer().setInputCol(c).setOutputCol(s"${c}_i") 
) 

val assembler = new VectorAssembler() 
    .setInputCols(featureColsIdx) 
    .setOutputCol("features") 

val slicer = new VectorSlicer() 
    .setInputCol("features") 
    .setOutputCol("string_features") 
    .setNames(featureColsIdx.init) 


val transformed = new Pipeline() 
    .setStages(indexers :+ assembler :+ slicer) 
    .fit(df) 
    .transform(df) 

pierwsze możemy wyodrębnić żądany metadane z cech:

val meta = transformed.select($"string_features") 
    .schema.fields.head.metadata 
    .getMetadata("ml_attr") 
    .getMetadata("attrs") 
    .getMetadataArray("nominal") 

i przekonwertować go na coś łatwiejszego korzystania

case class NominalMetadataWrapper(idx: Long, name: String, vals: Array[String]) 

// In general it could a good idea to make it a broadcast variable 
val lookup = meta.map(m => NominalMetadataWrapper(
    m.getLong("idx"), m.getString("name"), m.getStringArray("vals") 
)) 

Wreszcie mały UDF:

import scala.util.Try 

val transFeatures = udf((v: Vector) => lookup.map{ 
    m => Try(m.vals(v(m.idx.toInt).toInt)).toOption 
}) 

transformed.select(transFeatures($"string_features")).