根据值有效地跟踪字典的前 k 个键

Posted 2023-02-23

【中文标题】根据值有效地跟踪字典的前 k 个键

【英文标题】：Efficiently keeping track of the top-k keys of a dictionary based on value

【发布时间】：2013-03-03 19:32:35

【问题描述】：

当字典的键更新时，您如何有效地跟踪具有最大值的字典的前 k 个键？

我尝试过在每次更新后从字典中创建排序列表的简单方法（如 Getting key with maximum value in dictionary? 中所述），但是这种方法非常昂贵并且无法扩展。

现实世界的例子：

计算来自无限数据流的词频。在任何给定时刻，程序可能会被要求报告一个词是否在 current 最常见的值中。我们如何有效地做到这一点？

collections.Counter 太慢了

>>> from itertools import permutations
>>> from collections import Counter
>>> from timeit import timeit
>>> c = Counter()
>>> for x in permutations(xrange(10), 10):
    c[x] += 1


>>> timeit('c.most_common(1)', 'from __main__ import c', number=1)
0.7442058258093311
>>> sum(c.values())
3628800

计算这个值需要将近一秒钟！

我正在为 most_common() 函数寻找 O(1) 时间。这应该可以通过使用另一个数据结构来实现，该结构仅在内部存储当前的前 k 个项目，并跟踪当前的最小值。

【问题讨论】：

这似乎是试用heapq的绝佳机会。

@WeiYen 我已经想到了，但是如何增加已经在堆中的值呢？如果您存储所有项目，您可以轻松获得前 N 个最常见的，但是当值不断变化时，我看不出如何有效地更新计数。

您可以尝试类似于他们在heapq 文档中更新优先级的方式：docs.python.org/2/library/…

@WeiYen 有趣的是，我在文档中没有看到这一点（或者在我写答案时你的评论）。这种方法既好又简单，但在我们想要的恒定大小堆的情况下不起作用。另一方面，所有唯一词的堆会（a）变得巨大，（b）至少需要 O(N log N) 时间来找到 N 个最大的元素（另外你可能必须替换键） , 和 (c) 在构建完整计数器时包含 absurd 数量的“已删除”标记（堆将为您看到的 每个单词 有一个条目，而不仅仅是每个唯一的单词类型）。

您期望有多少个独特的词？也许可以利用计数永远不会减少的事实。要检查一个单词是否在前 N 个中，可能会更容易 - 你只需要找到任何 N 个较大的项目。

【参考方案1】：

我们可以实现一个跟踪 top-k 值的类，因为我不相信标准库有这个内置的。这将与主字典对象（可能是Counter）并行保持最新。您也可以将其用作主字典对象的子类的属性。

实施

class MostCommon(object):
    """Keep track the top-k key-value pairs.

    Attributes:
        top: Integer representing the top-k items to keep track of.
        store: Dictionary of the top-k items.
        min: The current minimum of any top-k item.
        min_set: Set where keys are counts, and values are the set of
            keys with that count.
    """
    def __init__(self, top):
        """Create a new MostCommon object to track key-value paris.

        Args:
            top: Integer representing the top-k values to keep track of.
        """
        self.top = top
        self.store = dict()
        self.min = None
        self.min_set = defaultdict(set)

    def _update_existing(self, key, value):
        """Update an item that is already one of the top-k values."""
        # Currently handle values that are non-decreasing.
        assert value > self.store[key]
        self.min_set[self.store[key]].remove(key)
        if self.store[key] == self.min:  # Previously was the minimum.
            if not self.min_set[self.store[key]]:  # No more minimums.
                del self.min_set[self.store[key]]
                self.min_set[value].add(key)
                self.min = min(self.min_set.keys())
        self.min_set[value].add(key)
        self.store[key] = value

    def __contains__(self, key):
        """Boolean if the key is one of the top-k items."""
        return key in self.store

    def __setitem__(self, key, value):
        """Assign a value to a key.

        The item won't be stored if it is less than the minimum (and
        the store is already full). If the item is already in the store,
        the value will be updated along with the `min` if necessary.
        """
        # Store it if we aren't full yet.
        if len(self.store) < self.top:
            if key in self.store:  # We already have this item.
                self._update_existing(key, value)
            else:  # Brand new item.
                self.store[key] = value
                self.min_set[value].add(key)
                if value < self.min or self.min is None:
                    self.min = value
        else:  # We're full. The value must be greater minimum to be added.
            if value > self.min:  # New item must be larger than current min.
                if key in self.store:  # We already have this item.
                    self._update_existing(key, value)
                else:  # Brand new item.
                    # Make room by removing one of the current minimums.
                    old = self.min_set[self.min].pop()
                    del self.store[old]
                    # Delete the set if there are no old minimums left.
                    if not self.min_set[self.min]:
                        del self.min_set[self.min]
                    # Add the new item.
                    self.min_set[value].add(key)
                    self.store[key] = value
                    self.min = min(self.min_set.keys())

    def __repr__(self):
        if len(self.store) < 10:
            store = repr(self.store)
        else:
            length = len(self.store)
            largest = max(self.store.itervalues())
            store = '<len=length, max=largest>'.format(length=length,
                                                           largest=largest)
        return ('self.__class__.__name__(top=self.top, min=self.min, '
                'store=store)'.format(self=self, store=store))

示例用法

>>> common = MostCommon(2)
>>> common
MostCommon(top=2, min=None, store=)
>>> common['a'] = 1
>>> common
MostCommon(top=2, min=1, store='a': 1)
>>> 'a' in common
True
>>> common['b'] = 2
>>> common
MostCommon(top=2, min=1, store='a': 1, 'b': 2)
>>> common['c'] = 3
>>> common
MostCommon(top=2, min=2, store='c': 3, 'b': 2)
>>> 'a' in common
False
>>> common['b'] = 4
>>> common
MostCommon(top=2, min=3, store='c': 3, 'b': 4)

更新值后的访问确实是O(1)

>>> counter = Counter()
>>> for x in permutations(xrange(10), 10):
        counter[x] += 1

>>> common = MostCommon(1)
>>> for key, value in counter.iteritems():
    common[key] = value

>>> common
MostCommon(top=1, min=1, store=(9, 7, 8, 0, 2, 6, 5, 4, 3, 1): 1)
>>> timeit('repr(common)', 'from __main__ import common', number=1)
1.3251570635475218e-05

访问是 O(1)，但是当在一个 O(n) 操作的 set-item 调用期间最小值发生变化时，其中n 是最高值的数量。这仍然比 Counter 好，每次访问都是 O(n)，其中 n 是整个词汇表的大小！

【参考方案2】：

collections.Counter.most_commondoes a pass over all the values, finding the N-th largest one by putting them in a heap as it goes（我认为是 O(M log N) 时间，其中 M 是字典元素的总数）。

heapq，正如 Wei Yen 在 cmets 中所建议的那样，可能工作正常：与字典平行，维护 N 个最大值中的 heapq，并在修改 dict 时检查该值是否在其中或现在应该在里面。问题是，正如您所指出的，接口实际上没有任何方法来修改已经 -现有元素。

您可以就地修改相关项目，然后运行heapq.heapify 以恢复heapiness。这需要对堆 (N) 的大小进行线性传递以找到相关项目（除非您正在做额外的簿记以将元素与位置相关联；可能不值得），以及另一个线性传递来重新堆化。如果某个元素不在列表中而现在在列表中，您需要通过替换最小元素将其添加到堆中（在线性时间内，除非有一些额外的结构）。

不过，heapq 私有接口包含一个函数 _siftdown，其中包含以下注释：

# 'heap' is a heap at all indices >= startpos, except possibly for pos.  pos
# is the index of a leaf with a possibly out-of-order value.  Restore the
# heap invariant.

听起来不错！调用 heapq._siftdown(heap, 0, pos_of_relevant_idx) 将在日志 N 时间修复堆。当然，您必须首先找到要递增的索引的位置，这需要线性时间。您可能会维护索引元素的字典以避免这种情况（也保留指向最小元素位置的指针），但是您要么必须复制 _siftdown 的源并修改它以更新字典当它交换东西时，或者做一个线性时间传递之后重建字典（但你只是试图避免线性传递......）。

要小心，这应该会在 O(log N) 时间内完成。但事实证明，有一个叫做Fibonacci heap 的东西确实支持您需要的所有操作，在（摊销的）常数 时间内。不幸的是，这是大 O 不是全部的情况之一。斐波那契堆的复杂性意味着在实践中，除了可能非常大的堆之外，它们实际上并不比二叉堆快。此外（也许是“因此”），虽然 Boost C++ 库确实包含一个，但我在快速谷歌搜索中没有找到标准的 Python 实现。

我首先尝试使用heapq，对您要更改的元素进行线性搜索，然后调用_siftdown；这是 O(N) 时间，与 Counter 方法的 O(M log N) 相比。如果结果太慢，您可以维护额外的索引字典并制作自己的_siftdown 版本来更新字典，这应该会花费 O(log N) 时间。如果这仍然太慢（我对此表示怀疑），您可以寻找 Boost 的 Fibonacci 堆（或其他实现）的 Python 包装器，但我真的怀疑这是否值得麻烦。

【参考方案3】：

使用collections.Counter 它已经为现实世界的例子做到了。您还有其他用例吗？

【讨论】：

出于某种原因，我认为Counter 对象无法使用其他数据进行更新，但查看文档似乎可以。谢谢！

我已经实现了这个，但是速度很慢。这是最有效的方法吗？