heapq --- 堆積佇列 (heap queue) 演算法

原始碼:Lib/heapq.py


這個模組實作了堆積佇列 (heap queue) 演算法,亦被稱為優先佇列 (priority queue) 演算法。

Heap(堆積)是一顆二元樹,樹上所有父節點的值都小於等於他的子節點的值。使用陣列實作,對於所有從0開始的 k 都滿足 heap[k] <= heap[2*k+1]heap[k] <= heap[2*k+2] 。為了比較節點的值,不存在的元素被視為無限大。heap 存在一個有趣的性質:樹上最小的元素永遠會在根節點 heap[0] 上。

下方的 API 跟一般教科書的 heap queue 演算法有兩個方面不同:第一,我們的索引從 0 開始計算,這會父節點與子節點之間的關係產生很微小的差異,但更符合 Python 從 0 開始索引的設計。第二,我們的 pop 方法會回傳最小的元素而不是最大的元素 ( 在教科書中被稱作 "min heap",而 "max heap" 因為他很適合做原地排序,所以更常出現在教科書中 )。

這兩個特性使得把 heap 當作一個標準的 Python list 檢視時不會出現意外:heap[0] 是最小的物件,heap.sort() 能保持 heap 的性質不變!

建立一個 heap 可以使用 list 初始化為 [],或者使用函式 heapify() 將一個已經有元素的 list轉成一個 heap。

此模組提供下面的函式

heapq.heappush(heap, item)

item 放進 heap,並保持 heap 性質不變。

heapq.heappop(heap)

heap 取出並回傳最小的元素,同時保持 heap 性質不變。如果 heap 是空的會產生 IndexError 錯誤。只存取最小元素但不取出可以使用 heap[0]

heapq.heappushpop(heap, item)

item 放入 heap ,接著從 heap 取出並回傳最小的元素。這個組合函式比呼叫 heappush() 之後呼叫 heappop() 更有效率。

heapq.heapify(x)

在線性時間內將 list x 轉為 heap,且過程不會申請額外記憶體。

heapq.heapreplace(heap, item)

heap 取出並回傳最小的元素,接著將新的 item 放進heap。heap 的大小不會改變。如果 heap 是空的會產生 IndexError 錯誤。

這個一次完成的操作會比呼叫 heappop() 之後呼叫 heappush() 更有效率,並在維護 heap 的大小不變時更為適當,取出/放入的組合函式一定會從 heap 回傳一個元素並用 item 取代他。

函式的回傳值可能會大於被加入的 item 。如果這不是你期望發生的,可以考慮使用 heappushpop() 替代,他會回傳 heap 的最小值和 item 兩個當中比較小的那個,並將大的留在 heap 內。

這個模組也提供三個利用 heap 實作的一般用途函式

heapq.merge(*iterables, key=None, reverse=False)

合併多個已排序的輸入並產生單一且已排序的輸出(舉例:合併來自多個 log 檔中有時間戳記的項目)。回傳一個 iterator 包含已經排序的值。

sorted(itertools.chain(*iterables)) 類似但回傳值是一個 iterable ,不會一次把所有資料都放進記憶體中,並且假設每一個輸入都已經(由小到大)排序過了。

有兩個選用參數,指定時必須被當作關鍵字參數指定。

key 參數指定了一個 key function 引數,用來從每一個輸入的元素中決定一個比較的依據。預設的值是 None (直接比較元素)。

reverse 是一個布林值,如果設定為 True ,則輸入的元素將以相反的比較順序進行合併。為了達成類似 sorted(itertools.chain(*iterables), reverse=True) 的行為,所有 iterables 必須由大到小排序。

在 3.5 版的變更: 加入選用參數 keyreverse

heapq.nlargest(n, iterable, key=None)

回傳一個包含資料 iterable 中前 n 大元素的 list 。如果有指定 key 引數,key 會是只有一個引數的函式,用來從每一個在 iterable 中的元素提取一個比較的依據(例如 key=str.lower )。效果相當於 sorted(iterable, key=key, reverse=True)[:n]

heapq.nsmallest(n, iterable, key=None)

回傳一個包含資料 iterable 中前 n 小元素的 list 。如果有指定 key 引數,key 會是只有一個引數的函式,用來從每一個在 iterable 中的元素提取一個比較的依據(例如 key=str.lower )。效果相當於 sorted(iterable, key=key)[:n]

後兩個函式在 n 值比較小時有最好的表現。對於較大的 n 值,只用 sorted() 函式會更有效率。同樣地,當 n==1 時,使用內建函式 min()max() 會有更好的效率。如果需要重複使用這些函式,可以考慮將 iterable 轉成真正的 heap 。

基礎範例

堆積排序 (heapsort) 可以透過將所有的值推入一個 heap,並且從 heap 中一個接一個彈出最小元素來實作:

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(len(h))]
...
>>> heapsort([1, 3, 5, 7, 9, 2, 4, 6, 8, 0])
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

雖然類似 sorted(iterable) ,但跟 sorted() 不同的是,這個實作不是 stable 的排序。

Heap 中的元素可以是 tuple 。這有利於將要比較的值(例如一個 task 的優先度)和主要資料放在一起排序。

>>> h = []
>>> heappush(h, (5, 'write code'))
>>> heappush(h, (7, 'release product'))
>>> heappush(h, (1, 'write spec'))
>>> heappush(h, (3, 'create tests'))
>>> heappop(h)
(1, 'write spec')

優先佇列實作細節

優先佇列 (priority queue) 是 heap 的常見用途之一,實作優先佇列伴隨著下列挑戰:

  • 排序的穩定性:如何將兩個擁有相同優先次序 (priority) 的 task 按照他們被加入的順序回傳?

  • Tuple的排序在某些情況下會壞掉,例如當 Tuple (priority, task) 的 priorities 相等且 tasks 沒有一個預設的排序時。

  • 當一個 heap 中 task 的 priority 改變時,如何將它移到 heap 正確的位置上?

  • 或者一個還沒被解決的 task 需要被刪除時,要如何從佇列中找到並刪除指定的 task?

一個針對前兩個問題的解法是:儲存一個包含 priority 、 entry count 和 task 三個元素的 tuple 。兩個 task 有相同 priority 時, entry count 會讓兩個 task 能根據加入的順序排序。因為沒有任何兩個 task 擁有相同的 entry count ,所以永遠不會直接使用 task 做比較。

task 無法比較的另一個解決方案是建立一個包裝器類別,該類別忽略 task 項目,只比較優先等級:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

剩下的問題可以藉由找到要刪除的 task 並更改它的 priority 或者直接將它移除。尋找一個 task 可以使用一個 dictionary 指向佇列當中的 entry 。

移除 entry 或更改它的 priority 更為困難,因為這會破壞 heap 的性質。所以一個可行的方案是將原本的 entry 做一個標記表示它已經被刪除,並新增一個擁有新的 priority 的 entry 。

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

原理

Heap 是一個陣列對於所有從0開始的 index k 都存在性質 a[k] <= a[2*k+1]a[k] <= a[2*k+2] 。為了方便比較,不存在的元素被視為無限大。Heap 的一個有趣的性質是:a[0] 永遠是最小的元素。

上述乍看之下有些奇怪的不變式,是為了實作一個對記憶體來說有效率的方法,其表示方式如同錦標賽一般。下列的數字為 k,而不是 a[k]

                               0

              1                                 2

      3               4                5               6

  7       8       9       10      11      12      13      14

15 16   17 18   19 20   21 22   23 24   25 26   27 28   29 30

在上面的樹當中,每個單元 k 都會位在 2*k+12*k+2 上方。如同體育賽事常見的錦標賽般,每個單元可視為其下方兩個單元當中的贏家,我們可以透過追溯整棵樹來找到該贏家曾經對戰過的所有對手。然而,在許多電腦應用中,我們不需要追溯贏家的完整對戰歷史。為了能更有效率地使用記憶體,當一個贏家晉級勝出時,我們用下方較低層級的另一個項目來取代它,至此規則變為一個單元以及它下方兩個單元,包含三個不同項目,但是最上方的單元「勝過」下方兩個單元。

如果能確保滿足這個 heap 的不變式,那麼索引 0 顯然是最終的贏家。移除並找到「下一個」贏家最簡單的演算法為:將一個輸家(例如上圖中的單元 30)移動到位置 0,然後從新的位置 0 不斷與下方的位置交換值來向下傳遞,直到滿足不變式為止。這個過程的複雜度顯然是樹的節點數目的對數級別。透過對所有項目疊代,可以得到一個複雜度為 O(n log n) 的排序。

這種排序有個好處,只要插入的項目沒有「贏過」你最後提取、索引為 0 的元素,你就可以在排序進行的同時有效率地插入新項目。這在模擬情境當中特別有用,其中樹能夠保存所有輸入事件,而「贏」意味著最小排程時間。當一個事件排程其它事件的執行時,因這些事件仍在等待進行,所以很容易將它們插入 heap 當中。因此, heap 是一個實現排程器的優秀資料結構(這就是我用以實作 MIDI 編曲器的方法 :-)。

多種用於實作排程器的結構現今已被廣泛研究,heap 對此非常有用,因為它們速度相當快,且速度幾乎不受其他因素影響,最壞情況與平均狀況差異無幾。也有其它整體說來更有效率的方法,然而它們的最壞情況可能會非常糟糕。

Heap 在為儲存於硬碟上的大量資料進行排序也非常有用。你可能已經知道,大量資料排序涉及 "runs" 的產生(也就是預先排序的序列,其大小通常與 CPU 記憶體的大小有關),之後再對這些 run 合併,而這些合併的過程通常相當巧妙 [1]。很重要的一點是,初始排序產生的 run 越長越好。錦標賽是達成這一點的好方法,若你用所有可用記憶體來舉行一場錦標賽,並透過替換與向下交換來處理所有適配當前 run 的值,那麼對於隨機產生的輸入,將可以產生長度兩倍於記憶體大小的 run。對於已模糊排序過的輸入,效果更好。

此外,若你將索引為 0 的項目輸出至磁碟,並取得一個無法適配當前錦標賽的輸入(因為該值「勝過」最後的輸出值),則該輸入值就無法插入至 heap 當中,因此 heap 的大小會減小。釋放出來的記憶體可以巧妙地立即再被運用,逐步建構出第二個 heap,其大小增加的速度會與第一個 heap 減少的速度一致。當第一個 heap 完全消失時,你可以切換至第二個 heap 開啟一個新 run 。這真是個聰明且相當有效率的做法!

總結來說,heap 是值得了解的有用記憶體結構。我在一些應用中使用它們,我認為能有一個 'heap' 模組是很棒的。:-)

註解