scala – 如何在Spark中转置RDD

如果N和M都很小,可以在内存中保存N×M个项目,那么使用RDD并不太有意义.但是转置它很容易：

val rdd = sc.parallelize(Seq(Seq(1,2,3),Seq(4,5,6),Seq(7,8,9)))
val transposed = sc.parallelize(rdd.collect.toSeq.transpose)

如果N或M大到不能在内存中保留N或M个条目,那么您不能拥有此大小的RDD行.在这种情况下,原始或转置的矩阵都不可能表示.

N和M可能是中等大小：您可以在内存中保留N或M个条目,但不能保留N×M个条目.在这种情况下,您必须炸毁矩阵并重新放在一起：

val rdd = sc.parallelize(Seq(Seq(1,9)))
// Split the matrix into one number per line.
val byColumnAndRow = rdd.zipWithIndex.flatMap {
  case (row,rowIndex) => row.zipWithIndex.map {
    case (number,columnIndex) => columnIndex -> (rowIndex,number)
  }
}
// Build up the transposed matrix. Group and sort by column index first.
val byColumn = byColumnAndRow.groupByKey.sortByKey().values
// Then sort by row index.
val transposed = byColumn.map {
  indexedRow => indexedRow.toSeq.sortBy(_._1).map(_._2)
}