高效负载CSV坐标格式(COO)输入到本地矩阵spark

val df = spark.read.format("csv").option("header", "true").load("path/to/csv/file.csv")

import org.apache.spark.ml.linalg.{SparseVector, Vectors}

val cooMatrix = df.rdd.map(row => {
  val rowIndex = row.getAs[Int]("row_index")
  val colIndex = row.getAs[Int]("col_index")
  val value = row.getAs[Double]("value")
  (rowIndex, colIndex, value)
}).collect()

val numRows = cooMatrix.map(_._1).max + 1
val numCols = cooMatrix.map(_._2).max + 1

val localMatrix = Array.ofDim[Double](numRows, numCols)

cooMatrix.foreach { case (rowIndex, colIndex, value) =>
  localMatrix(rowIndex)(colIndex) = value
}

val cooMatrixRDD = spark.sparkContext.parallelize(cooMatrix)
val localMatrixRDD = spark.sparkContext.parallelize(localMatrix)