在pyspark中重新划分密集矩阵答案

【问题标题】：repartition a dense matrix in pyspark在pyspark中重新划分密集矩阵
【发布时间】：2016-04-21 01:54:17
【问题描述】：

我在 pyspark 中有一个密集矩阵（100*100），我想将repartition 分成十组，每组包含 10 行。

from pyspark import SparkContext, SparkConf
from pyspark.mllib import *
sc = SparkContext("local", "Simple App")
dm2 = Matrices.dense(100, 100, RandomRDDs.uniformRDD(sc, 10000).collect())
newRdd = sc.parallelize(dm2.toArray())
rerdd = newRdd.repartition(10)

上面的代码导致rerdd 包含100 个元素。我想将此矩阵 dm2 呈现为按行分区的块（例如，分区中有 10 行）。

【问题讨论】：

我尝试使用普通的numpy矩阵并转换为RDD并在此处对其进行分区。 stackoverflow.com/questions/36739687/…

标签： python apache-spark pyspark

【解决方案1】：

我没有多大意义，但你可以例如做这样的事情

mat =  Matrices.dense(100, 100, np.arange(10000))

n_par = 10
n_row = 100

rdd = (sc
    .parallelize(
        # Add indices
        enumerate(
            # Extract and reshape values
            mat.values.reshape(n_row, -1)))
    # Partition and sort by row index
    .repartitionAndSortWithinPartitions(n_par, lambda i: i // n_par))

检查分区数和每个分区的行数：

rdd.glom().map(len).collect()
## [10, 10, 10, 10, 10, 10, 10, 10, 10, 10

检查第一行是否包含所需的数据：

assert np.all(rdd.first()[1] == np.arange(100))

【讨论】：

rdd.getNumPartitions() 返回 100
对我来说很好用。