如何在 Spark 中将 maptype 转换为 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 类型。 例如 SparkML VectorAssembler 允许组合多个列（所有数字类型、布尔类型或向量类型）。

编辑：

到目前为止，我的解决方案是将地图单独分解为列，然后使用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：

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

Answer 1

据我所知，Spark 中没有内置方法，因此在这种情况下，UDF 将是一个合适的解决方案。这是一个使用Map[String, Double] 的列并返回一个机器学习向量：

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

由于 Map 没有顺序，因此也不保证生成的向量具有特定顺序。

示例输入 (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，就像在问题编辑中一样。

---

编辑：

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

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

然后适当更改UDF 以考虑keys 的顺序，默认值为0.0：

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