用DataFrame代替textFile的Spark Mllib kmeans示例

Question

我正在尝试在一个浮点数的数据框上运行Spark（1.3.1）Mllib k-means聚类。

我正在遵循Spark提供的聚类示例。

https://spark.apache.org/docs/1.3.1/mllib-clustering.html

然而，我使用的不是文本文件，而是由一个双精度列组成的数据框（为了简单起见）。根据Mllib文档，我需要将其转换为Kmeans函数的向量。到目前为止，我有以下代码：

    import org.apache.spark.mllib.linalg.Vectors
    val parsedData = data.map(s => Vectors.dense(s(0))).cache()

我收到了错误

error: overloaded method value dense with alternatives:
(values: Array[Double])org.apache.spark.mllib.linalg.Vector and
(firstValue: Double,otherValues: Double*)org.apache.spark.mllib.linalg.Vector
cannot be applied to (Any)
val parsedData = sample2.map(s => Vectors.dense(s(1))).cache()
                                          ^

有更好的方法吗？

我已经阅读了类似的帖子，但我觉得不够相似： 如何将已知结构的RDD转换为向量 以及这篇文章如何将org.apache.spark.rdd.RDD [Array[Double]]转换为Spark MLlib所需的Array[Double] 它处理文本数据

- Michael Plazzer

---

因此，这个问题存在于在Hadoop集群上运行Spark的背景下，其中使用Hive查询了一个表。因此产生了dataframe。我怀疑随着更多的组织转向Hadoop，这将成为越来越常见的情况。 - Michael Plazzer

Answer 1

由于import org.apache.spark.sql.Row可以存储任何类型的值，因此它的apply方法具有以下签名：

 def apply(i: Int): Any 

Vectors.dense 期望参数为 Double。有几种处理方式。假设你想从列 x 中提取值。首先，你可以在 Row 构造函数上进行模式匹配：

data.select($"x").map {
    case  Row(x: Double) => Vectors.dense(x)
}

如果您更喜欢位置方法，可以使用模式匹配来匹配从apply返回的值：

data.select($"x").map(row => row(0) match {
    case x: Double => Vectors.dense(x)
})

最终，您可以使用 toDouble 方法：

data.select($"x").map(r => Vectors.dense(r.getDouble(0)))

select 部分是可选的，但它使得在行上进行模式匹配更容易，并且可以保护您免受一些天真错误（例如将错误索引传递给 get）。

如果您想要提取大量列，逐个提取可能会很繁琐。 在这种情况下，可以使用以下方法：

data.select($"x", $"y", $"z").map(r => Vectors.dense(
    r.toSeq.map({ case col: Double => col }).toArray)
)

- zero323

---

在第一种情况下，你不会这样写吗：_parsedData = data.map { case Row(x: Double, _) => Vectors.dense(x) }_？否则我会得到一个“类型不匹配”的错误。 - Michael Plazzer

你确定不是 scala.MatchError 吗？如果是这样，我已经更新了一个答案来避免在整行匹配时出现问题。 - zero323

Answer 2

什么情况：

val parsedData = data.rdd.map(s => Vectors.dense(s.getDouble(0))).cache()

如果数据是一个包含单列双精度浮点数的数据框，则此方法可行。 如果您的数据框中有多列，则只需添加更多的获取操作，例如：

val parsedData = data.rdd.map(s => Vectors.dense(s.getDouble(0),s.getDouble(1))).cache()

- Dr VComas

---

谢谢您，医生。这正是我想要做的。 - Michael Plazzer