如何在Spark中将映射类型转换为SparkML稀疏矢量？

Question

我的原始模式包含许多我想在ML模型中使用的映射类型，因此我需要将它们转换为SparkML稀疏向量。

root
 |-- colA: map (nullable = true)
 |    |-- key: string
 |    |-- value: double (valueContainsNull = true)
 |-- colB: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- colC: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)

背景：SparkML模型要求将数据形成特征向量。有一些实用程序可以生成特征向量，但都不支持maptype类型。e、 g.SparkML VectorAssembler允许组合多个列（所有数字类型、布尔类型或向量类型）。

编辑：

到目前为止，我的解决方案是将映射分解为单独的列，然后使用向量汇编程序：

val listkeysColA = df.select(explode($"colA"))
  .select($"key").as[Int].distinct.collect.sorted

val exploded= df.select(listkeysColA.map(x => 
  $"colA".getItem(x).alias(x.toString)): _*).na.fill(0) 

val columnNames = exploded.columns

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

编辑2：

我应该补充一点，我的地图中的数据非常稀疏，事先没有已知的键集。这就是为什么在我当前的解决方案中，我首先对data进行一次传递以收集和排序键。然后我使用getItem（keyName）访问值。

Answer 1

据我所知，Spark中没有用于此的内置方法，因此在这种情况下，UDF将是合适的解决方案。这里是一个采用映射[字符串，双精度]的列并返回ML向量的方法：

val toVector = udf((m: Map[String, Double]) => Vectors.dense(m.values.toArray).toSparse)

由于映射没有阶，因此也不能保证生成的向量具有特定阶。

示例输入（df）：

+---------------------------------+---------------------------------+
|colA                             |colB                             |
+---------------------------------+---------------------------------+
|Map(a -> 1.0, b -> 2.0, c -> 3.0)|Map(a -> 1.0, b -> 2.0, c -> 3.0)|
+---------------------------------+---------------------------------+

并使用UDF，

val df2 = df.withColumn("colA", toVector($"colA")).withColumn("colB", toVector($"colB"))

给出以下输出：

+-------------+-------------+
|colA         |colB         |
+-------------+-------------+
|[1.0,2.0,3.0]|[1.0,2.0,3.0]|
+-------------+-------------+

其中两列都是向量类型。

root
 |-- colA: vector (nullable = true)
 |-- colB: vector (nullable = true)

如果您想将所有列合并到一个向量中，这里可以使用向量汇编程序（VectorAssembler），如问题编辑中所示。

编辑：

如果您需要保持值的特定顺序，那么您需要像以前一样首先收集所有键。但是，您可以避免使用爆炸：

val keys = df.select($"colA")
  .flatMap(_.getAs[Map[String, Int]]("colA").keys)
  .distinct
  .collect
  .sorted

然后适当更改自定义项，以考虑键的顺序，默认值为0.0：

val toVector = udf((m: Map[String, Double]) => 
  Vectors.dense(keys.map(key => m.getOrElse(key, 0.0))).toSparse
)