Áp dụng IndexToString cho các tính năng vectơ trong Spark

Question

Bối cảnh: Tôi có một khung dữ liệu trong đó tất cả các giá trị phân loại đã được lập chỉ mục bằng cách sử dụng 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") 
} 

Sau đó, tôi đã sử dụng VectorAssembler để vector hóa tất cả các cột tính năng (bao gồm các cột danh mục được lập chỉ mục).

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

Sau khi áp dụng trình phân loại và một vài bước bổ sung, tôi kết thúc với khung dữ liệu có nhãn, tính năng và dự đoán. Tôi muốn mở rộng tính năng vector của mình để tách các cột để chuyển đổi các giá trị được lập chỉ mục trở lại dạng Chuỗi ban đầu của chúng.

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

Câu hỏi: Có đơn giản cách để làm điều này, hoặc là cách tiếp cận tốt nhất để bằng cách nào đó gắn cột dự đoán vào khung dữ liệu thử nghiệm?

Những gì tôi đã cố gắng:

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

Áp dụng điều này mang lại cho tôi các cột mà tôi muốn, nhưng họ đang có phong độ Vector (vì nó có nghĩa là để làm) và không gõ đúp.

Edit: Các đầu ra mong muốn là khung ban đầu dữ liệu (ví dụ: tính năng phân loại như String không index) với một cột bổ sung cho thấy nhãn dự đoán (mà trong trường hợp của tôi là 0 hoặc 1).

Ví dụ, nói đầu ra của phân loại của tôi trông như thế này:

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

Bằng cách áp dụng VectorSlicer trên mỗi tính năng tôi sẽ nhận được:

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

Mà là tuyệt vời, nhưng tôi cần:

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

Để sau đó có thể sử dụng IndexToString và chuyển đổi thành:

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

hoặc thậm chí:

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

Answer 1

Vâng, nó không phải là một hoạt động rất hữu ích nhưng nó cần được thể trích xuất thông tin cần thiết sử dụng siêu dữ liệu cột và UDF đơn giản.Tôi giả sử dữ liệu của bạn đã được tạo ra một đường ống dẫn tương tự như này:

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) 

Đầu tiên chúng ta có thể trích xuất siêu dữ liệu mong muốn từ các tính năng:

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

và chuyển nó sang một cái gì đó dễ dàng hơn để sử dụng

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

Cuối cùng là một UDF nhỏ:

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