2013-11-26 14 views
6

我一直非常沮丧,python radix的许多实现都在Web上进行排序。推基数排序(和python)到其极限

它们始终使用10的基数,并通过除以10的幂或取数字的log10来获得它们迭代的数字的数字。这是非常低效的,因为与比特移位相比,log10并不是特别快速的操作,比位移快了近100倍!

更高效的实现使用256的基数并逐字节地对数字进行排序。这允许使用可靠的快速位操作符来完成所有'字节获取'。不幸的是,似乎绝对没有人在Python中使用位运算符而不是对数来实现基数排序。

于是,我带着问题到我自己的手,并用此兽,它在上排序小数组大约一半的速度运行,那么快运行在较大的想出了(如len周围10,000,000):

import itertools 

def radix_sort(unsorted): 
    "Fast implementation of radix sort for any size num." 
    maximum, minimum = max(unsorted), min(unsorted) 

    max_bits = maximum.bit_length() 
    highest_byte = max_bits // 8 if max_bits % 8 == 0 else (max_bits // 8) + 1 

    min_bits = minimum.bit_length() 
    lowest_byte = min_bits // 8 if min_bits % 8 == 0 else (min_bits // 8) + 1 

    sorted_list = unsorted 
    for offset in xrange(lowest_byte, highest_byte): 
     sorted_list = radix_sort_offset(sorted_list, offset) 

    return sorted_list 

def radix_sort_offset(unsorted, offset): 
    "Helper function for radix sort, sorts each offset." 
    byte_check = (0xFF << offset*8) 

    buckets = [[] for _ in xrange(256)] 

    for num in unsorted: 
     byte_at_offset = (num & byte_check) >> offset*8 
     buckets[byte_at_offset].append(num) 

    return list(itertools.chain.from_iterable(buckets)) 

这个版本的基数排序的工作原理是找出它必须排序的字节(如果你只传递256以下的整数,它将只排序一个字节等),然后通过转储排序LSB中的每个字节装入桶中,然后将桶连在一起。对需要排序的每个字节重复此操作,并且您在O(n)时间内有很好的排序数组。

但是,它并没有像它可能的那么快,而且我希望能够把它写得更快,然后再把它写成更好的基数排序,而不是像所有其他基数排序那样。

运行cProfile这个告诉我,很多的时间被消耗在append方法列表,这让我觉得此块:

for num in unsorted: 
     byte_at_offset = (num & byte_check) >> offset*8 
     buckets[byte_at_offset].append(num) 

radix_sort_offset是吃了大量的时间。这也是一个阻碍,如果你真的看看它,整个工作的90%。这段代码看起来好像可能是numpy -ized,我认为这会带来相当的性能提升。不幸的是,我对numpy的更复杂的功能并不擅长,所以一直无法弄清楚。帮助将非常感激。

我目前使用itertools.chain.from_iterable来平整buckets,但如果有人有更快的建议,我相信它也会有帮助。

最初,我有一个get_byte函数返回一个数字的第012个字节,但内联代码给了我一个巨大的速度提升,所以我做到了。

其他有关实施或其他方面的更多性能的评论也值得赞赏。我想听到任何事情和你所拥有的一切。

回答

9

你已经意识到,

for num in unsorted: 
    byte_at_offset = (num & byte_check) >> offset*8 
    buckets[byte_at_offset].append(num) 

是大多数时间的推移 - 好;-)

有超速之类的话两个标准的招数,都具有移动不变办out of loops:

  1. 计算循环外部的“offset * 8”。将其存储在局部变量中。每次迭代保存一个乘法。
  2. 在循环外部添加bucketappender = [bucket.append for bucket in buckets]。保存每次迭代的方法查找。

它们组合起来,并且循环的样子:

for num in unsorted: 
    bucketappender[(num & byte_check) >> ofs8](num) 

它折叠到一个语句也节省了对本地vrbl店/取每次迭代操作码。

但是,在更高层次上,加速基数排序的标准方法是使用更大的基数。 256有什么神奇的?没有什么,除此之外,它是方便位移的。但512,1024,2048也是如此......这是一种经典的时间/空间折衷。

PS:对于很长的数字,

(num >> offset*8) & 0xff 

将运行得更快。这是因为您的num & byte_check需要的时间与log(num)成比例 - 通常需要创建一个大约为num的整数。

+1

好东西。这导致了非常强大的加速,并允许这种基数排序在一个列表10,000,000长的基础上以4096的基数排序,但是这确实使得它在短列表上令人尴尬地执行。编辑:刚才意识到你是写timsort的人。先生,我的帽子是给你的。 – reem

+1

赫 - 我敢打赌,你在这个列表中没有任何负数的整数;-)基数排序非常好,但是当你超越非负数整数时,这个小小的摆弄就会变得更加棘手。顺便说一句,我写了Python的'list.sort()',我并不感到冒犯你的速度更快:-) –

0

这是一个古老的线程,但当我看到基数排列正整数数组时,我遇到了这个问题。我试图看看我能否比已经非常快速的timsort做得更好(帽子再次给你,Tim Peters),它实现了python内建的排序和排序!要么我不明白上述代码的某些方面,或者如果我这样做,上述代码有一些问题恕我直言。

  1. 它只对从最小项目的最高字节开始,以最大项目的最高字节结束的字节进行排序。在某些特殊数据的情况下,这可能是可以的。但总的来说,这种方法无法区分因低位而不同的项目。例如:

    arr=[65535,65534] 
    radix_sort(arr) 
    

    产生错误输出:

    [65535, 65534] 
    
  2. 在辅助函数用来循环的范围是不正确的。我的意思是,如果lowest_byte和highest_byte是相同的,那么辅助函数的执行就完全被跳过了。顺便说一下,我必须在2个地方改变xrange范围。

  3. 经过修改,以解决上述2点,我得到它的工作。但是它需要花费10-20倍于Python内置的排序或排序!我知道timsort是非常有效的,并利用数据中已经排序的运行。但我试图看看是否可以使用先前的知识,我的数据在我的排序中都是正整数。为什么基数排序与timsort相比做得非常糟糕?我使用的数组大小约为80K项。是否因为除了算法效率之外的timsort实现还有其他效率可能源于可能使用低级库?或者我完全错过了一些东西?我用修改后的代码如下:

    import itertools 
    
    def radix_sort(unsorted): 
        "Fast implementation of radix sort for any size num." 
        maximum, minimum = max(unsorted), min(unsorted) 
    
        max_bits = maximum.bit_length() 
        highest_byte = max_bits // 8 if max_bits % 8 == 0 else (max_bits // 8) + 1 
    
    # min_bits = minimum.bit_length() 
    # lowest_byte = min_bits // 8 if min_bits % 8 == 0 else (min_bits // 8) + 1 
    
        sorted_list = unsorted 
    # xrange changed to range, lowest_byte deleted from the arguments 
        for offset in range(highest_byte): 
         sorted_list = radix_sort_offset(sorted_list, offset) 
    
        return sorted_list 
    
    def radix_sort_offset(unsorted, offset): 
        "Helper function for radix sort, sorts each offset." 
        byte_check = (0xFF << offset*8) 
    
    # xrange changed to range 
        buckets = [[] for _ in range(256)] 
    
        for num in unsorted: 
         byte_at_offset = (num & byte_check) >> offset*8 
         buckets[byte_at_offset].append(num) 
    
        return list(itertools.chain.from_iterable(buckets))