将基数排序（和 python）推向极限

Question

我对网络上的许多 python 基数排序实现感到非常沮丧。

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

一个更有效的实现使用 256 的基数并按字节对数字进行排序。这允许使用非常快速的位运算符来完成所有“字节获取”。不幸的是，似乎绝对没有人在 python 中实现了使用位运算符而不是对数的基数排序。

所以，我把事情掌握在自己的手中，想出了这个野兽，它在小型数组上的运行速度大约是 sorted 的一半，而在大型数组上的运行速度几乎相同（例如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 函数，它返回一个数字的 nth 字节，但内联代码给了我巨大的速度提升，所以我做到了。

还感谢任何其他有关实现或挤出更多性能的方法的 cmets。我想听听你所拥有的一切。

Answer 1

这是一个旧线程，但我在寻找基数排序正整数数组时遇到了这个问题。我试图看看我是否能比已经非常快的 timsort 做得更好（再次向你致敬，Tim Peters），它实现了 python 的内置排序和排序！要么我不理解上述代码的某些方面，要么如果我理解，那么上面提供的代码有一些问题恕我直言。

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

   arr=[65535,65534]
   radix_sort(arr)
   

   产生错误的输出：

   [65535, 65534]
   

2. 用于循环辅助函数的范围不正确。我的意思是，如果最低字节和最高字节相同，则完全跳过辅助函数的执行。顺便说一句，我不得不将 xrange 更改为 2 个位置。

3. 通过修改以解决上述 2 点，我得到了它的工作。但它花费的时间是 python 内置 sorted 或 sort 的 10-20 倍！我知道 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))
   

Answer 2

您可以简单地使用现有的 C 或 C++ 实现之一，例如 例如，integer_sort 来自 Boost.Sort 或 u4_sort 来自 usort。从 Python 调用原生 C 或 C++ 代码非常容易，请参阅How to sort an array of integers faster than quicksort?

我完全理解你的沮丧。虽然已经2年多了，numpy still does not have radix sort。我会让 NumPy 开发人员知道他们可以简单地获取现有实现之一；许可应该不是问题。

Answer 3

你已经意识到了

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

是大部分时间都去的地方 - 很好;-)

有两个标准技巧可以加速这种事情，都与将不变量移出循环有关：

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 一样大的整数。