Scala下Spark RDD的50×50矩阵不匹配错误及CSV导出问题

我来帮你一步步解决这两个问题：

1. 修复矩阵不匹配错误 #

这个错误最常见的原因是输入RDD[Vector]中的向量维度不一致（比如有的向量长度不是50），因为Statistics.corr要求所有输入向量的维度必须完全相同才能计算相关矩阵。

第一步：排查维度一致性 #

先运行这段代码检查你的输入RDD：

// 假设你的输入RDD名为inputRDD: RDD[Vector]
val dimensionStats = inputRDD.map(_.size).distinct().collect()
if (dimensionStats.length != 1) {
  println(s"发现不一致的向量维度：${dimensionStats.mkString(", ")}")
} else if (dimensionStats.head != 50) {
  println(s"向量维度不符合预期：预期50，实际${dimensionStats.head}")
}

第二步：修复维度问题 #

如果发现维度不一致，你需要预处理数据：

• 过滤掉维度不符合的向量：

  val filteredRDD = inputRDD.filter(_.size == 50)
  

• 或者补全/截断向量到50维（根据你的业务需求选择）：

  // 截断到50维（如果向量长度超过50）
  val truncatedRDD = inputRDD.map(vec => Vectors.dense(vec.toArray.take(50)))
  // 补0到50维（如果向量长度不足50）
  val paddedRDD = inputRDD.map(vec => {
    val arr = vec.toArray
    val fixedArr = if (arr.length < 50) arr ++ Array.fill(50 - arr.length)(0.0) else arr
    Vectors.dense(fixedArr)
  })
  

额外检查：确保corr调用正确 #

确认你是用正确的方式计算相关矩阵：

// 计算皮尔逊相关矩阵（也可以替换为"spearman"）
val correlMatrix = Statistics.corr(filteredRDD, "pearson")

这里的filteredRDD必须是RDD[Vector]类型，每个向量代表一个样本，向量的每个元素对应一个特征（这样50个特征就会输出50×50的矩阵）。

2. 将相关矩阵导出到CSV文件 #

由于50×50的矩阵属于小数据，不需要分布式处理，直接在Driver端操作即可，这里提供两种常用方法：

方法一：写入本地文件（适用于本地模式或Driver节点可访问的路径） #

import java.io.PrintWriter

// 将Matrix转换为CSV格式字符串
val csvContent = correlMatrix.rowIter
  .map(row => row.toArray.mkString(",")) // 每行元素用逗号分隔
  .mkString("\n") // 行与行用换行分隔

// 写入到本地指定路径
new PrintWriter("./correlation_matrix.csv") {
  write(csvContent)
  close()
}

方法二：写入分布式存储（如HDFS，适用于集群模式） #

如果需要写入HDFS或其他分布式存储，可以把CSV内容转为RDD再保存：

// 将矩阵每行转为字符串，生成RDD[String]
val csvRows = spark.sparkContext.parallelize(
  correlMatrix.rowIter.map(_.toArray.mkString(",")).toSeq
)

// 保存到分布式路径（替换为你的集群实际路径）
csvRows.saveAsTextFile("hdfs://your-cluster-path/correlation_matrix")

注意：saveAsTextFile会生成多个分区文件，如果需要单个文件，可以先调用coalesce(1)再保存，但小数据量下影响不大。

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