如何使用Spark的repartitionAndSortWithinPartitions？

我正在尝试构建一个最小的工作示例repartitionAndSortWithinPartitions以便理解该函数。到目前为止我已经完成了（不起作用，不同的值会乱七八糟地乱七八糟）

def partval(partID:Int, iter: Iterator[Int]): Iterator[Tuple2[Int, Int]] = {
  iter.map( x => (partID, x)).toList.iterator
}

val part20to3_chaos = sc.parallelize(1 to 20, 3).distinct
val part20to2_sorted = part20to3_chaos.repartitionAndSortWithinPartitions(2)
part20to2_sorted.mapPartitionsWithIndex(partval).collect

但出现错误

Name: Compile Error
Message: <console>:22: error: value repartitionAndSortWithinPartitions is not a member of org.apache.spark.rdd.RDD[Int]
             val part20to2_sorted = part20to3_chaos.repartitionAndSortWithinPartitions(2)

我尝试使用scaladoc，但无法找到哪个类提供repartitionAndSortWithinPartitions。 （顺便说一句：这个 scaladoc 并不令人印象深刻：为什么MapPartitionsRDD丢失的？我如何寻找方法？）

意识到我需要一个分区对象，接下来我尝试

val rangePartitioner = new org.apache.spark.RangePartitioner(2, part20to3_chaos)
val part20to2_sorted = part20to3_chaos.repartitionAndSortWithinPartitions(rangePartitioner)
part20to2_sorted.mapPartitionsWithIndex(partval).collect

but got

Name: Compile Error
Message: <console>:22: error: type mismatch;
 found   : org.apache.spark.rdd.RDD[Int]
 required: org.apache.spark.rdd.RDD[_ <: Product2[?,?]]
Error occurred in an application involving default arguments.
         val rPartitioner = new org.apache.spark.RangePartitioner(2, part20to3_chaos)

我如何让它编译？请问我可以得到一个工作示例吗？

你的问题是part20to3_chaos is an RDD[Int], while OrderedRDDFunctions.repartitionAndSortWithinPartitions是一种运行在RDD[(K, V)], where K是关键并且V是值。

repartitionAndSortWithinPartitions首先会重新分区基于提供的分区器的数据，然后按键排序:

/**
 * Repartition the RDD according to the given partitioner and, 
 * within each resulting partition, sort records by their keys.
 *
 * This is more efficient than calling `repartition` and then sorting within each partition
 * because it can push the sorting down into the shuffle machinery.
 */
def repartitionAndSortWithinPartitions(partitioner: Partitioner): RDD[(K, V)] = 
  self.withScope {
    new ShuffledRDD[K, V, V](self, partitioner).setKeyOrdering(ordering)
}

所以看起来这并不完全是您要找的。

如果你想要一个普通的旧排序，你可以使用sortBy，因为它不需要密钥：

scala> val toTwenty = sc.parallelize(1 to 20, 3).distinct
toTwenty: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at distinct at <console>:33

scala> val sorted = toTwenty.sortBy(identity, true, 3).collect
sorted: Array[Int] = 
    Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20)

你经过的地方sortBy顺序（升序或降序）以及要创建的分区数量。