如何转的RDD在星火

Question

我有一个RDD这样的：

1 2 3 
4 5 6 
7 8 9 

这是一个矩阵。现在我想转置这样的RDD：

1 4 7 
2 5 8 
3 6 9 

我该怎么做？

Answer 1

假设您有一个N × M矩阵。

如果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, 2, 3), Seq(4, 5, 6), Seq(7, 8, 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) 
} 

Answer 2

的第一份草案，而无需使用收集（），所以一切运行者侧，并没有什么在驱动程序进行：

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

rdd.flatMap(row => (row.map(col => (col, row.indexOf(col))))) // flatMap by keeping the column position 
    .map(v => (v._2, v._1)) // key by column position 
    .groupByKey.sortByKey // regroup on column position, thus all elements from the first column will be in the first row 
    .map(_._2)    // discard the key, keep only value 

该解决方案的问题在于，如果在分布式系统中执行操作，转置矩阵中的列将最终混洗。会想到一个改进版本

我的想法是，除了将'列号'附加到矩阵的每个元素，我们还附加'行号'。因此，我们可以按列位置键和按键重新分组，然后我们可以对行号中的每一行重新排序，然后从结果中去除行数/列数。 我只是没有办法在将文件导入RDD时知道行号。

您可能认为将矩阵列和行号附加到每个矩阵元素很重要，但我想这是付出代价才有可能以分布式方式处理您的输入，从而处理巨大矩阵。

当我找到排序问题的解决方案时，将更新答案。

Answer 3

从Spark 1.6开始，您可以使用DataFrame上的pivot operation，具体取决于数据的实际形状，如果将其放入DF中，您可以将列转换为行，以下databricks blog非常有用，因为它详细描述一些代码示例的透视用例