如何在Spark中将RowMatrix转换为Array或将其导出为CSV？

Question

我有这个Scala代码：

val mat: CoordinateMatrix = new CoordinateMatrix(data)
val rowMatrix: RowMatrix = mat.toRowMatrix()

val svd: SingularValueDecomposition[RowMatrix, Matrix] = rowMatrix.computeSVD(100, computeU = true)

val U: RowMatrix = svd.U // The U factor is a RowMatrix.
val S: Vector = svd.s // The singular values are stored in a local dense vector.
val V: Matrix = svd.V // The V factor is a local dense matrix.

val uArray: Array[Double] = U.toArray // doesn't work, because there is not toArray function in RowMatrix type
val sArray: Array[Double] = S.toArray // works good
val vArray: Array[Double] = V.toArray // works good

我该如何将 U 转换为 uArray 或类似类型，以便将其打印到 CSV 文件中？

- Leszek Malinowski

Answer 1

这是一个基本操作，考虑到U是一个RowMatrix，下面是你需要做的：

val U = svd.U

rows()是一种RowMatrix方法，它允许您通过行从RowMatrix获取RDD。

您只需要在RowMatrix上应用rows，映射RDD [Vector]以创建一个数组，然后将其连接成一个字符串创建RDD [String]即可。

val rdd = U.rows.map( x => x.toArray.mkString(","))

现在您所需要做的就是保存RDD：

rdd.saveAsTextFile(path)

- eliasah

---

不知道为什么，但是saveAsTexFile并没有像我预期的那样工作。然而，使用你的提示和rows()函数，我最终成功导出了它。谢谢 :) - Leszek Malinowski

你期望些什么？ - eliasah

我以为我会收到一个可以在Excel中查看的.txt、.csv或类似的文件。但它却创建了一个以文件路径中最后一个字符串命名的文件夹，然后抛出了NullPointerException异常。我做了一个"快速修复"，我将在回答中发布它。 - Leszek Malinowski

是的，saveAsTextFile会写入分区，因此您无法获得txt文件或csv文件。我通常执行cat命令将文件夹中的文件连接到一个csv文件中，例如cat test/part-* > test.csv。 - eliasah

Answer 2

它有效：

def exportRowMatrix(matrix:RDD[String], fileName: String) = {
  val pw = new PrintWriter(fileName)
  matrix.collect().foreach(line => pw.println(line))

  pw.flush
  pw.close
}

val rdd = U.rows.map( x => x.toArray.mkString(","))
exportRowMatrix(rdd, "U.csv")

- Leszek Malinowski

---

collect方法的唯一问题是，如果矩阵太大，它会给你一个Java堆错误，所以要小心！ - eliasah

@eliasah 哇！我已经遇到了Java堆错误，所以现在我知道原因了。我在Spark设置中增加了内存，问题就解决了 ;) - Leszek Malinowski

增加内存并不总是一个好的解决方案，但如果它对你有用，那就很好。 - eliasah