陣列排序是一個常見的操作。基於比較的排序演算法其效能下限是O(nlog(n))，但在分散式環境下面我們可以併發，從而提高效能。這裡展示了Spark中陣列排序的實現，並分析了效能，同時嘗試找到導致效能提升的原因。

官方示例

import sys
 
from pyspark import SparkContext
 
if __name__ == "__main__":
    if len(sys.argv) != 2:
        print >> sys.stderr, "Usage: sort <file>"
        exit(-1)
    sc = SparkContext(appName="PythonSort")
    lines = sc.textFile(sys.argv[1], 1)
    sortedCount = lines.flatMap(lambda x: x.split(' ')) \
        .map(lambda x: (int(x), 1)) \
        .sortByKey(lambda x: x)
    # This is just a demo on how to bring all the sorted data back to a single node.
    # In reality, we wouldn't want to collect all the data to the driver node.
    output = sortedCount.collect()
    for (num, unitcount) in output:
        print num
 
    sc.stop()
 

方法textFile讀入一個文字檔案，並在Spark環境裡建立相應的RDD集。這個資料集存放在lines變數中。方法flatMap和map不同，map返回的是一個key,value

>>> rdd = sc.parallelize([2, 3, 4])
>>> sorted(rdd.flatMap(lambda x: range(1, x)).collect())
[1, 1, 1, 2, 2, 3]

容易看出，這段程式實際上只是呼叫了Spark提供的排序介面sortByKey，而不是在程式碼中實現一個排序演算法。方法sortByKey的底層實現如下：

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.size)      
	: RDD[(K, V)] =  {    
	val part = new RangePartitioner(numPartitions, self, ascending)    
	new ShuffledRDD[K, V, V](self, part)      
		.setKeyOrdering(if (ascending) ordering else ordering.reverse)  
}

訪問本人部落格獲得更多資訊：magic01