Applicazione di IndexToString al vettore di funzioni in Spark

Question

Contesto: Ho un frame di dati in cui tutti i valori categoriali sono stati indicizzati utilizzando 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") 
} 

poi ho usato VectorAssembler vectorize tutte le colonne funzionalità (tra cui quelli categorici indicizzati).

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

Dopo l'applicazione del classificatore e pochi passaggi aggiuntivi alla fine con un frame di dati che ha un'etichetta, caratteristiche, e la previsione. Mi piacerebbe espandere il vettore delle mie funzionalità per separare le colonne per convertire i valori indicizzati nella loro forma String originale.

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

Domanda: C'è una semplice modo di fare questo , o è l'approccio migliore per attaccare in qualche modo la colonna previsione al telaio dati di test?

Quello che ho cercato:

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

Applicando questo mi dà le colonne che voglio, ma sono in forma vettoriale (in quanto è destinato a fare) e non tipo double.

Edit: L'uscita desiderata è il frame originale di dati (cioè funzioni categoriali come stringa non indice) con una colonna aggiuntiva che indica l'etichetta prevista (che nel mio caso è 0 o 1).

Per esempio, dicono che l'uscita del mio classificatore sembrava qualcosa di simile:

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

Applicando VectorSlicer su ogni caratteristica vorrei avere:

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

Che è grande, ma ho bisogno di:

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

per poi essere in grado di utilizzare IndexToString e convertirlo in:

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

o anche:

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

Answer 1

Beh, non è un'operazione molto utile ma dovrebbe essere possibile estrarre le informazioni richieste utilizzando colonna di metadati e di semplice UDF.Presumo i dati è stata creata una pipeline di simile a questo:

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) 

Prima possiamo estrarre i metadati desiderato dalle caratteristiche:

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

e convertirlo in qualcosa di più facile da usare

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") 
)) 

Infine una piccola 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")).