python - 如何有效地过滤具有任意长度元组作为键的字典？

Question

长话短说

为具有可变维度键的字典实现过滤功能的最有效方法是什么？过滤器应该采用与字典键相同维度的元组，并输出字典中与过滤器匹配的所有键，使得 filter[i] 为 None 或 filter[i] == key[i] 对于所有维度 i。

---

在我当前的项目中，我需要处理包含大量数据的字典。字典的一般结构是这样的，它包含以 2 到 4 个整数作为键和整数作为值的元组。字典中的所有键都具有相同的维度。为了说明，以下是我需要处理的字典示例:

{(1, 2): 1, (1, 5): 2}
{(1, 5, 3): 2}
{(5, 2, 5, 2): 8}

这些词典包含很多词条，最大的大约有 20 000 个词条。我经常需要过滤这些条目，但通常只查看关键元组的某些索引。理想情况下，我希望有一个函数可以提供一个过滤器元组。然后该函数应返回与过滤器元组匹配的所有键。如果过滤器元组包含一个 None 条目，那么它将匹配该索引处字典键元组中的任何值。

函数应该为具有二维键的字典做什么的示例:

>>> dict = {(1, 2): 1, (1, 5): 2, (2, 5): 1, (3, 9): 5}
>>> my_filter_fn((1, None))
{(1, 2), (1, 5)}
>>> my_filter_fn((None, 5))
{(1, 5), (2, 5)}
>>> my_filter_fn((2, 4))
set()
>>> my_filter_fn((None, None))
{(1, 2), (1, 5), (2, 5), (3, 9)}

由于我的字典有不同的元组维度，我尝试通过编写一个考虑元组维度的生成器表达式来解决这个问题:

def my_filter_fn(entries: dict, match: tuple):
    return (x for x in entries.keys() if all(match[i] is None or match[i] == x[i]
                                             for i in range(len(key))))

不幸的是，与完全手动写出条件相比，这非常慢 ((match[0] is None or match[0] === x[0]) and (match[1] is None or match[1] == x[1]); 对于 4 个维度，这大约慢了 10 倍。这对我来说是个问题，因为我需要经常进行这种过滤。

以下代码演示了性能问题。仅提供代码来说明问题并启用测试的重现。您可以跳过代码部分，结果如下。

import random
import timeit


def access_variable_length():
    for key in entry_keys:
        for k in (x for x in all_entries.keys() if all(key[i] is None or key[i] == x[i]
                                                       for i in range(len(key)))):
            pass


def access_static_length():
    for key in entry_keys:
        for k in (x for x in all_entries.keys() if
                  (key[0] is None or x[0] == key[0])
                  and (key[1] is None or x[1] == key[1])
                  and (key[2] is None or x[2] == key[2])
                  and (key[3] is None or x[3] == key[3])):
            pass


def get_rand_or_none(start, stop):
    number = random.randint(start-1, stop)
    if number == start-1:
        number = None
    return number


entry_keys = set()
for h in range(100):
    entry_keys.add((get_rand_or_none(1, 200), get_rand_or_none(1, 10), get_rand_or_none(1, 4), get_rand_or_none(1, 7)))
all_entries = dict()
for l in range(13000):
    all_entries[(random.randint(1, 200), random.randint(1, 10), random.randint(1, 4), random.randint(1, 7))] = 1

variable_time = timeit.timeit("access_variable_length()", "from __main__ import access_variable_length", number=10)
static_time = timeit.timeit("access_static_length()", "from __main__ import access_static_length", number=10)

print("variable length time: {}".format(variable_time))
print("static length time: {}".format(static_time))

结果:

variable length time: 9.625867042849316
static length time: 1.043319165662158

我想避免必须创建三个不同的函数 my_filter_fn2、my_filter_fn3 和 my_filter_fn4 来涵盖我的词典的所有可能维度和然后使用静态尺寸过滤。我知道可变维度的过滤总是比固定维度的过滤慢，但我希望它不会慢近 10 倍。由于我不是 Python 专家，我希望有一种巧妙的方法可以重新制定我的可变维度生成器表达式，从而获得更好的性能。

以我描述的方式过滤庞大字典的最有效方法是什么？

Answer 1

感谢有机会思考集合和字典中的元组。这是 Python 的一个非常有用和强大的角落。

Python 是解释型的，因此如果您来自编译语言，一个好的经验法则是尽可能避免复杂的嵌套迭代。如果您正在编写复杂的 for 循环或推导式，那么总是值得考虑是否有更好的方法来完成它。

列表下标 (stuff[i]) 和 range (len(stuff)) 在 Python 中效率低下且冗长，很少需要。迭代更有效(也更自然):

for item in stuff:
    do_something(item)

以下代码速度很快，因为它使用了 Python 的一些优势:理解、字典、集合和元组解包。

有迭代，但它们简单而肤浅。 整段代码只有一条if语句，每次过滤操作只执行4次。这也有助于提高性能，并使代码更易于阅读。

方法的解释...

原始数据中的每个键:

{(1, 4, 5): 1}

按位置和值索引:

{
    (0, 1): (1, 4, 5),
    (1, 4): (1, 4, 5),
    (2, 5): (1, 4, 5)
}

(Python 从零开始对元素进行编号。)

索引被整理成一个由元组集组成的大查找字典:

{
    (0, 1): {(1, 4, 5), (1, 6, 7), (1, 2), (1, 8), (1, 4, 2, 8), ...}
    (0, 2): {(2, 1), (2, 2), (2, 4, 1, 8), ...}
    (1, 4): {(1, 4, 5), (1, 4, 2, 8), (2, 4, 1, 8), ...}
    ...
}

一旦构建了这个查找(并且构建得非常高效)，过滤就是设置交集和字典查找，两者都快如闪电。即使是大型字典，过滤也需要几微秒。

该方法处理元数为 2、3 或 4(或任何其他元数)的数据，但 arity_filtered() 仅返回成员数与过滤器元组相同的键。因此，此类为您提供了将所有数据一起过滤或分别处理不同大小的元组的选项，在性能方面几乎没有选择。

大型随机数据集(11,500 个元组)的计时结果是构建查找需要 0.30 秒，100 次查找需要 0.007 秒。

from collections import defaultdict
import random
import timeit


class TupleFilter:
    def __init__(self, data):
        self.data = data
        self.lookup = self.build_lookup()

    def build_lookup(self):
        lookup = defaultdict(set)
        for data_item in self.data:
            for member_ref, data_key in tuple_index(data_item).items():
                lookup[member_ref].add(data_key)
        return lookup

    def filtered(self, tuple_filter):
        # initially unfiltered
        results = self.all_keys()
        # reduce filtered set
        for position, value in enumerate(tuple_filter):
            if value is not None:
                match_or_empty_set = self.lookup.get((position, value), set())
                results = results.intersection(match_or_empty_set)
        return results

    def arity_filtered(self, tuple_filter):
        tf_length = len(tuple_filter)
        return {match for match in self.filtered(tuple_filter) if tf_length == len(match)}

    def all_keys(self):
        return set(self.data.keys())


def tuple_index(item_key):
    member_refs = enumerate(item_key)
    return {(pos, val): item_key for pos, val in member_refs}


data = {
    (1, 2): 1,
    (1, 5): 2,
    (1, 5, 3): 2,
    (5, 2, 5, 2): 8
}

tests = {
     (1, 5): 2,
     (1, None, 3): 1,
     (1, None): 3,
     (None, 5): 2,
}

tf = TupleFilter(data)
for filter_tuple, expected_length in tests.items():
    result = tf.filtered(filter_tuple)
    print("Filter {0} => {1}".format(filter_tuple, result))
    assert len(result) == expected_length
# same arity filtering
filter_tuple = (1, None)
print('Not arity matched: {0} => {1}'
      .format(filter_tuple, tf.filtered(filter_tuple)))
print('Arity matched: {0} => {1}'
      .format(filter_tuple, tf.arity_filtered(filter_tuple)))
# check unfiltered results return original data set
assert tf.filtered((None, None)) == tf.all_keys()


>>> python filter.py
Filter (1, 5) finds {(1, 5), (1, 5, 3)}
Filter (1, None, 3) finds {(1, 5, 3)}
Filter (1, None) finds {(1, 2), (1, 5), (1, 5, 3)}
Filter (None, 5) finds {(1, 5), (1, 5, 3)}
Arity filtering: note two search results only: (1, None) => {(1, 2), (1, 5)}