如何将列转换为向量类型？答案

【问题标题】：How to convert column to vector type?如何将列转换为向量类型？
【发布时间】：2016-03-18 01:23:06
【问题描述】：

我在 Spark 中有一个 RDD，其中的对象基于案例类：

ExampleCaseClass(user: User, stuff: Stuff)

我想使用 Spark 的 ML 管道，因此我将其转换为 Spark 数据帧。作为管道的一部分，我想将其中一列转换为条目为向量的列。由于我希望该向量的长度随模型而变化，因此应将其作为特征转换的一部分内置到管道中。

所以我尝试如下定义一个 Transformer：

class MyTransformer extends Transformer {

  val uid = ""
  val num: IntParam = new IntParam(this, "", "")

  def setNum(value: Int): this.type = set(num, value)
  setDefault(num -> 50)

  def transform(df: DataFrame): DataFrame = {
    ...
  }

  def transformSchema(schema: StructType): StructType = {
    val inputFields = schema.fields
    StructType(inputFields :+ StructField("colName", ???, true))
  }

  def copy (extra: ParamMap): Transformer = defaultCopy(extra)

}

如何指定结果字段的数据类型（即填写???）？它将是某个简单类（Boolean、Int、Double 等）的 Vector。看起来 VectorUDT 可能有效，但这是 Spark 私有的。由于任何 RDD 都可以转换为 DataFrame，因此任何案例类都可以转换为自定义 DataType。但是我不知道如何手动进行这种转换，否则我可以将它应用到一些包装向量的简单案例类中。

此外，如果我为列指定向量类型，当我去拟合模型时，VectorAssembler 是否会正确地将向量处理为单独的特征？

对 Spark 尤其是 ML Pipeline 来说还是新手，因此感谢任何建议。

【问题讨论】：

1) Vectors in Spark 仅支持 Double 类型（这些与 Scala 不同 Vector, 请参阅 stackoverflow.com/q/31255756/1560062）2）VectorUDT is not private 3）并非每个 RDD 都可以转换为 DataFrame，或者至少不能直接转换，也不是每个案例类都可以自动使用 asd Dataset (Frame) 元素。
并不是所有的RDDs都可以转换成DataFrames。 Dirty RDD
4) 如果我为列指定向量类型，VectorAssembler 是否会正确地将向量处理为单独的特征 - 如果不明白问题所在。汇编程序只是连接列（如果存在，则元数据）。
我真正想做的是创建更多的列（不仅仅是一列中的长向量），但我不知道如何做到这一点，而且效率极低。我应该只需要遍历一次数据来创建所有这些列，而现在我只能看到如何使用循环来完成。
这可能解释了我对 VectorAssembler 的评论，因为我真的希望该向量的每个元素都是它自己的特征。

标签： scala apache-spark apache-spark-ml

【解决方案1】：

import org.apache.spark.ml.linalg.SQLDataTypes.VectorType  
def transformSchema(schema: StructType): StructType = {
  val inputFields = schema.fields
  StructType(inputFields :+ StructField("colName", VectorType, true))
}

在 spark 2.1 VectorType 中，VectorUDT 公开可用：

package org.apache.spark.ml.linalg

import org.apache.spark.annotation.{DeveloperApi, Since}
import org.apache.spark.sql.types.DataType

/**
 * :: DeveloperApi ::
 * SQL data types for vectors and matrices.
 */
@Since("2.0.0")
@DeveloperApi
object SQLDataTypes {

  /** Data type for [[Vector]]. */
  val VectorType: DataType = new VectorUDT

  /** Data type for [[Matrix]]. */
  val MatrixType: DataType = new MatrixUDT
}

【讨论】：

【解决方案2】：

import org.apache.spark.mllib.linalg.{Vector, Vectors}

case class MyVector(vector: Vector)
val vectorDF = Seq(
  MyVector(Vectors.dense(1.0,3.4,4.4)),
  MyVector(Vectors.dense(5.5,6.7))
).toDF

vectorDF.printSchema
root
 |-- vector: vector (nullable = true)

println(vectorDF.schema.fields(0).dataType.prettyJson)
{
  "type" : "udt",
  "class" : "org.apache.spark.mllib.linalg.VectorUDT",
  "pyClass" : "pyspark.mllib.linalg.VectorUDT",
  "sqlType" : {
    "type" : "struct",
    "fields" : [ {
      "name" : "type",
      "type" : "byte",
      "nullable" : false,
      "metadata" : { }
    }, {
      "name" : "size",
      "type" : "integer",
      "nullable" : true,
      "metadata" : { }
    }, {
      "name" : "indices",
      "type" : {
        "type" : "array",
        "elementType" : "integer",
        "containsNull" : false
      },
      "nullable" : true,
      "metadata" : { }
    }, {
      "name" : "values",
      "type" : {
        "type" : "array",
        "elementType" : "double",
        "containsNull" : false
      },
      "nullable" : true,
      "metadata" : { }
    } ]
  }
}

【讨论】：