【问题标题】:How can I filter a dictionary with arbitrary length tuples as keys efficiently?如何有效地过滤具有任意长度元组作为键的字典?
【发布时间】:2017-11-02 06:37:34
【问题描述】:

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))))

不幸的是,与完全手动写出条件相比,这相当慢((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))

结果:

变长时间:9.625867042849316
静态时长:1.043319165662158

我希望避免创建三个不同的函数my_filter_fn2my_filter_fn3my_filter_fn4 来覆盖我的字典的所有可能维度,然后使用静态维度过滤。我知道过滤可变尺寸总是比过滤固定尺寸慢,但希望它不会慢近 10 倍。由于我不是 Python 专家,我希望有一种聪明的方法可以重新制定我的可变维度生成器表达式,以提供更好的性能。

按照我描述的方式过滤庞大字典的最有效方法是什么?

【问题讨论】:

  • 对你的对象使用内置名称是不好的,所以filter应该重命名(例如filter_entries
  • @AzatIbrakov 谢谢,我改了。

标签: python performance dictionary filtering


【解决方案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)}

【讨论】:

  • 谢谢,好主意。不幸的是,在实践中,我的字典时常变化。我需要检查重建查找表的成本是否被查找加速所抵消。
  • 首先,构建查找字典比串行搜索要快,那么为什么不直接重新构建呢?其次,没有什么能阻止您向查找字典添加新数据项。如果有删除,可以通过根据数据检查过滤结果来轻松处理它们。 Udates 更棘手,并且最直接地由查找重建处理
  • 您介意我将它包含在 MIT 许可的软件包中吗?
【解决方案2】:

我做了一些修改:

  • 你不需要使用dict.keys方法来遍历key,遍历dict对象本身就会给我们它的key,

  • 创建了单独的模块,有助于阅读和修改:

    • preparations.py 带有用于生成测试数据的助手:

      import random
      
      left_ends = [200, 10, 4, 7]
      
      
      def generate_all_entries(count):
          return {tuple(random.randint(1, num)
                        for num in left_ends): 1
                  for _ in range(count)}
      
      
      def generate_entry_keys(count):
          return [tuple(get_rand_or_none(1, num)
                        for num in left_ends)
                  for _ in range(count)]
      
      
      def get_rand_or_none(start, stop):
          number = random.randint(start - 1, stop)
          if number == start - 1:
              number = None
          return number
      
    • functions.py 测试功能,
    • main.py 用于基准测试。
  • 将参数传递给函数而不是从全局范围获取它们,因此给定的静态和可变长度版本变为

    def access_static_length(all_entries, entry_keys):
        for key in entry_keys:
            for k in (x
                      for x in all_entries
                      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 access_variable_length(all_entries, entry_keys):
        for key in entry_keys:
            for k in (x
                      for x in all_entries
                      if all(key[i] is None or key[i] == x[i]
                             for i in range(len(key)))):
                pass
    
  • timeit.repeat 的结果上使用min 而不是timeit.timeit 以获得最具代表性的结果(更多信息请参见this answer),

  • entries_keys 元素计数从10 更改为100(包括结束),步骤为10

  • all_entries 元素计数从10000 更改为15000(包括结束),步骤为500


但回到正题。

改进

  1. 我们可以通过跳过对键中具有None 值的索引的检查来改进过滤

    def access_variable_length_with_skipping_none(all_entries, entry_keys):
        for key in entry_keys:
            non_none_indexes = {i
                                for i, value in enumerate(key)
                                if value is not None}
            for k in (x
                      for x in all_entries.keys()
                      if all(key[i] == x[i]
                             for i in non_none_indexes)):
                pass
    
  2. 下一个建议是使用numpy:

    import numpy as np
    
    
    def access_variable_length_numpy(all_entries, entry_keys):
        keys_array = np.array(list(all_entries))
        for entry_key in entry_keys:
            non_none_indexes = [i
                                for i, value in enumerate(entry_key)
                                if value is not None]
            non_none_values = [value
                               for i, value in enumerate(entry_key)
                               if value is not None]
            mask = keys_array[:, non_none_indexes] == non_none_values
            indexes, _ = np.where(mask)
            for k in map(tuple, keys_array[indexes]):
                pass
    

基准测试

main.py的内容:

import timeit
from itertools import product

number = 5
repeat = 10
for all_entries_count, entry_keys_count in product(range(10000, 15001, 500),
                                                   range(10, 101, 10)):
    print('all entries count: {}'.format(all_entries_count))
    print('entry keys count: {}'.format(entry_keys_count))
    preparation_part = ("from preparation import (generate_all_entries,\n"
                        "                         generate_entry_keys)\n"
                        "all_entries = generate_all_entries({all_entries_count})\n"
                        "entry_keys = generate_entry_keys({entry_keys_count})\n"
                        .format(all_entries_count=all_entries_count,
                                entry_keys_count=entry_keys_count))
    static_time = min(timeit.repeat(
        "access_static_length(all_entries, entry_keys)",
        preparation_part + "from functions import access_static_length",
        repeat=repeat,
        number=number))
    variable_time = min(timeit.repeat(
        "access_variable_length(all_entries, entry_keys)",
        preparation_part + "from functions import access_variable_length",
        repeat=repeat,
        number=number))
    variable_time_with_skipping_none = min(timeit.repeat(
        "access_variable_length_with_skipping_none(all_entries, entry_keys)",
        preparation_part +
        "from functions import access_variable_length_with_skipping_none",
        repeat=repeat,
        number=number))
    variable_time_numpy = min(timeit.repeat(
        "access_variable_length_numpy(all_entries, entry_keys)",
        preparation_part +
        "from functions import access_variable_length_numpy",
        repeat=repeat,
        number=number))

    print("static length time: {}".format(static_time))
    print("variable length time: {}".format(variable_time))
    print("variable length time with skipping `None` keys: {}"
          .format(variable_time_with_skipping_none))
    print("variable length time with numpy: {}"
          .format(variable_time_numpy))

在我的机器上使用 Python 3.6.1 给出:

all entries count: 10000
entry keys count: 10
static length time: 0.06314293399918824
variable length time: 0.5234129569980723
variable length time with skipping `None` keys: 0.2890012050011137
variable length time with numpy: 0.22945181500108447
all entries count: 10000
entry keys count: 20
static length time: 0.12795891799760284
variable length time: 1.0610534609986644
variable length time with skipping `None` keys: 0.5744297259989253
variable length time with numpy: 0.5105678180007089
all entries count: 10000
entry keys count: 30
static length time: 0.19210158399801003
variable length time: 1.6491422000035527
variable length time with skipping `None` keys: 0.8566724129996146
variable length time with numpy: 0.7363859869983571
all entries count: 10000
entry keys count: 40
static length time: 0.2561357790000329
variable length time: 2.08878050599742
variable length time with skipping `None` keys: 1.1256247100027394
variable length time with numpy: 1.0066140279996034
all entries count: 10000
entry keys count: 50
static length time: 0.32130833200062625
variable length time: 2.6166040710013476
variable length time with skipping `None` keys: 1.4147321179989376
variable length time with numpy: 1.1700750320014777
all entries count: 10000
entry keys count: 60
static length time: 0.38276188999952865
variable length time: 3.153736616997776
variable length time with skipping `None` keys: 1.7147898039984284
variable length time with numpy: 1.4533947029995034
all entries count: 10000
entry keys count: 70
...
all entries count: 15000
entry keys count: 80
static length time: 0.7141444490007416
variable length time: 6.186657476999244
variable length time with skipping `None` keys: 3.376506028998847
variable length time with numpy: 3.1577993860009883
all entries count: 15000
entry keys count: 90
static length time: 0.8115685330012639
variable length time: 7.14327938399947
variable length time with skipping `None` keys: 3.7462387939995097
variable length time with numpy: 3.6140603050007485
all entries count: 15000
entry keys count: 100
static length time: 0.8950150890013902
variable length time: 7.829741768000531
variable length time with skipping `None` keys: 4.1662235900003
variable length time with numpy: 3.914334102999419

简历

我们可以看到numpy 的版本不如预期的那么好,这似乎不是numpy 的错。

如果我们删除将过滤后的数组记录转换为 tuples 和 map 并离开

for k in keys_array[indexes]:
    ...

那么它会非常快(比静态长度版本快),所以问题在于从numpy.ndarray对象到tuple的转换。

过滤掉None 输入键给我们带来了大约 50% 的速度增益,所以请随意添加。

【讨论】:

    【解决方案3】:

    假设你有一本字典 - d

    d = {(1,2):3,(1,4):5,(2,4):2,(1,3):4,(2,3):6,(5,1):5,(3,8):5,(3,6):9}
    

    首先你可以得到字典键-

    keys = d.keys()
    =>
    dict_keys([(1, 2), (3, 8), (1, 3), (2, 3), (3, 6), (5, 1), (2, 4), (1, 4)])
    

    现在让我们定义一个函数is_match,它可以根据您的条件决定给定两个元组是否相等-
    is_match((1,7),(1,None))is_match((1,5),(None,5))is_match((1,4),(1,4)) 将返回Trueis_match((1,7),(1,8))is_match((4,7),(6,12)) 将返回 False

    def if_equal(a, b):
        if a is None or b is None:
            return True
        else:
            if a==b:
                return True
            else:
                return False
    
    is_match = lambda a,b: False not in list(map(if_equal, a, b))
    
    tup = (1, None)
    matched_keys = [key for key in keys if is_match(key, tup)]
    =>
    [(1, 2), (1, 3), (1, 4)]
    

    【讨论】:

    • 谢谢,但是这个解决方案的性能甚至比我已经提出的更差(参见“可变长度”下的问题)。
    【解决方案4】:

    我没有一个漂亮的答案,但这种优化通常会使代码更难阅读。但是,如果您只需要更快的速度,您可以做两件事。

    首先,我们可以直接消除循环内部的重复计算。您说每个字典中的所有条目都具有相同的长度,因此您可以计算一次,而不是在循环中重复。这对我来说减少了大约 20%:

    def access_variable_length():
        try:
            length = len(iter(entry_keys).next())
        except KeyError:
            return
        r = list(range(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 r)):
                pass
    

    不漂亮,我同意。但是我们可以通过使用eval 构建固定长度函数来使其更快(甚至更丑!)。像这样:

    def access_variable_length_new():
        try:
            length = len(iter(entry_keys).next())
        except KeyError:
            return
        func_l = ["(key[{0}] is None or x[{0}] == key[{0}])".format(i) for i in range(length)]
        func_s = "lambda x,key: " + " and ".join(func_l)
        func = eval(func_s)
        for key in entry_keys:
            for k in (x for x in all_entries.keys() if func(x,key)):
                pass
    

    对我来说,这几乎和静态版本一样快。

    【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2019-10-17
    • 1970-01-01
    • 1970-01-01
    • 2011-08-04
    • 1970-01-01
    • 1970-01-01
    • 2023-03-18
    相关资源
    最近更新 更多