如何在Spark中转置RDD

我有这样的RDD:

1 2 3 4 5 6 7 8 9 

这是一个matrix。 现在我想转置这样的RDD:

 1 4 7 2 5 8 3 6 9 

我该怎么做?

假设你有一个N×M的matrix。

如果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线。 在这种情况下,原始matrix或转置matrix是不可能表示的。

N和M可以是中等大小:您可以在内存中保存N或M个条目,但不能保存N×M个条目。 在这种情况下,你必须炸毁matrix并重新组合在一起:

 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) } 

没有使用collect()的初稿,所以一切都运行在工作人员的一方,没有任何事情在驱动程序上完成:

 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 

这个解决scheme的问题是,如果在分布式系统中执行操作,转置matrix中的列将最终被洗牌。 会想到一个改进的版本

我的想法是,除了将“列号”附加到matrix的每个元素之外,我们还附加“行号”。 所以我们可以按列的位置键和按键重新分组,然后我们可以对行号中的每一行重新sorting,然后从结果中去掉行/列号。 在将文件导入RDD时,我只是无法知道行号。

你可能会认为把一个列和一个行号附加到每个matrix元素上是很重的,但是我想这是付出的代价来有可能以分布式的方式处理你的input,从而处理巨大的matrix。

当我findsorting问题的解决scheme时,将更新答案。

从Spark 1.6开始,可以使用DataFrame上的pivot操作 ,具体取决于数据的实际形状,如果将其放入DF中,则可以将列转换为行,但下面的databricks博客非常有用,因为它详细描述了一个数字使用代码示例来转换用例