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

Question

TL; DR

使用可变维度键实现字典过滤功能的最有效方法是什么？过滤器应采用与字典键相同尺寸的元组,并输出字典中与过滤器匹配的所有键,以便filter[i] is None or 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))))

不幸的是,与完全用hand((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覆盖我的字典的所有可能的维度,然后使用静态维度过滤.我知道对于可变尺寸的过滤总是比对固定尺寸的过滤慢,但希望它不会慢几十倍.因为我不是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)}