Spark Mllib kmeans 示例，使用数据框而不是 textFile

Posted 2023-03-12

【中文标题】Spark Mllib kmeans 示例，使用数据框而不是 textFile

【英文标题】：Spark Mllib kmeans example, with a dataframe instead of textFile

【发布时间】：2015-10-01 06:28:31

【问题描述】：

我正在尝试在浮点数数据帧上运行 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()
                                          ^

有更好的方法吗？

我读过这个类似的帖子，但我觉得它不够相似： How to turn a known structured RDD to Vector 还有这个How to convert org.apache.spark.rdd.RDD[Array[Double]] to Array[Double] which is required by Spark MLlib 处理文本数据

【问题讨论】：

所以这个问题存在于 Spark 在 Hadoop 集群上运行的上下文中，其中使用 Hive 查询了一个表。因此数据框。我怀疑随着越来越多的组织迁移到 Hadoop，这将成为一种越来越普遍的情况。

【参考方案1】：

怎么样：

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

如果 data 是单列双精度的数据框，这应该可以工作。 如果您的数据框中有更多列，则只需添加更多获取，例如：

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

【讨论】：

谢谢医生，这正是我想做的。

【参考方案2】：

由于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 部分是可选的，但它可以更轻松地对 Row 进行模式匹配，并保护您免受一些幼稚的错误，例如将错误的索引传递给 get。

如果想要逐一提取大量列可能会很麻烦。在这种情况下，这样的事情可能会很有用：

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

【讨论】：

在第一种情况下，你会不会写： parsedData = data.map case Row(x :Double, _) => Vectors.dense(x) ？否则我会收到“类型不匹配”错误。

确定不是scala.MatchError？如果是这样，我已经更新了答案以避免匹配整行的问题。