如何将RDD[Matrix[Double]]转换为RDD[Vector]（矩阵每行转Vector）

核心思路是对RDD内每个Matrix实例做行迭代，再通过flatMap算子将所有行展开为独立的Vector元素，所有转换逻辑均在Executor端分布式执行，无需拉取全量数据到Driver节点。

现有单Matrix转RDD[Vector]的方案通常需要将Matrix数据拉取到Driver端再并行化生成RDD，不适用于分布式的RDD[Matrix]场景。本方案无Shuffle开销，支持超大规模数据集处理。

最简实现（兼容稠密/稀疏矩阵） #

Spark的Matrix接口原生提供rowIter方法可直接返回矩阵所有行的迭代器，元素类型为Vector，直接配合flatMap即可完成转换：

// 导入mllib包依赖，若使用ml包可调整导入路径为org.apache.spark.ml.linalg下的对应类
import org.apache.spark.mllib.linalg.{Matrix, Vector}
import org.apache.spark.rdd.RDD

// 你的原始RDD[Matrix[Double]]数据集
val sourceRDD: RDD[Matrix] = _

val resultRDD: RDD[Vector] = sourceRDD.flatMap(_.rowIter)

可选：保留原矩阵关联元数据 #

如果需要保留每行Vector所属的原矩阵标识，可在转换时携带对应元数据：

// 示例：为每个Vector添加所属原矩阵的唯一ID标识
val resultWithMetaRDD: RDD[(Long, Vector)] = sourceRDD.zipWithIndex()
  .flatMap { case (matrix, matrixId) =>
    matrix.rowIter.map(rowVec => (matrixId, rowVec))
  }

注意事项 #

• 若使用Spark ML包下的矩阵类，仅需将导入路径替换为org.apache.spark.ml.linalg.Matrix和org.apache.spark.ml.linalg.Vector即可，转换逻辑完全一致
• 若矩阵行数量差异较大，可在转换完成后调用repartition算子调整分区数，避免后续作业出现数据倾斜

内容的提问来源于stack exchange，提问作者Alain ux