将多个过滤器应用于元组列表

Question

我正在寻找一种有效的pythonic方法,将多个过滤器应用于元组列表.

举个例子,假设这样的过滤器:

def f1(t): return t[3]<10
def f2(t): return t[0]!=1
def f3(t): return t[1] in ("lisa","eric")
def f4(t): return t[3]>2

像这样的n元组(即db-records):

tuples=[
(0,'tom','...',8),
(1,'john','...',17),
(2,'lisa','...',1),
(3,'eric','...',18)
]

以下作品:

def nFilter(filters,tuples):
    if filters and tuples:
        return nFilter(filters,filter(filters.pop(),tuples))
    else: return tuples

结果如下:

>>> nFilter([f1,f2,f3],tuples)
[(2, 'lisa', '...', 1)]

和

>>> nFilter([f1,f2,f3,f4],tuples)
[]

但我想知道是否有更直接的方式; 我想到的是功能组合(即f1(f2(...fn(tuples)...))),用于任意功能列表.在文档中引用了包含函数的函数库compose,但链接都已死亡.

此外,由于我计划在相当大的数据集上使用它,并且可能在生产Web服务中使用大量过滤器,因此它必须是高效的,并且我无法确切地说这个解决方案是否正确.

欢迎任何建议或改进.

Answer 1

改进:用迭代替换递归

实际上并没有"任意函数列表的组合函数"; 但是,使用简单的for循环构建过滤器链非常容易:

def nFilter(filters, tuples):
    for f in filters:
        tuples = filter(f, tuples)
    return tuples

改进:通过限制和速度订购过滤器

链式迭代器的速度非常快,总运行时间往往会受到谓词函数调用的支配.

通过对谓词进行排序以最小化总工作量,可以获得最佳结果.一般来说,最好在昂贵的测试之前安排廉价测试,并在不过滤掉许多情况的测试之前进行更严格的测试.

例

在这个例子中,谓词具有大约相同的成本(函数调用,元组索引,并且相比于恒定),但它们在限制程度而变化(在t[2]==4过滤器出80的箱子%,而t[0]>1与t[1]<3每一个唯一的过滤器出口40 %的数据).

>>> from itertools import product

>>> filters = [lambda t: t[2]==4, lambda t: t[0]>1, lambda t: t[1]<3]
>>> for tup in nFilter(filters, product(range(5), repeat=3)):
        print(tup)

(2, 0, 4)
(2, 1, 4)
(2, 2, 4)
(3, 0, 4)
(3, 1, 4)
(3, 2, 4)
(4, 0, 4)
(4, 1, 4)
(4, 2, 4)

注释中提到了注释

• 当输入iterable为空时,过滤器函数对谓词进行零应用.这就像在空列表上执行for循环一样.

• 每个过滤器都会减少输入封闭过滤器的数据量.因此,每个过滤器仅被应用于通过先前过滤器的数据.

• 不要担心lambda示例中的内容.它具有与常规功能相同的功能def.它只是编写过滤器列表的便捷方式.

• 在Python 3中,filter()函数已更新为返回迭代器而不是列表.在Python 2中,您可以使用itertools.ifilter()而不是filter()来实现相同的效果.

Answer 2

你在找这样的东西吗？

filters = (f1,f2,f3,f4)
filtered_list = filter( lambda x: all(f(x) for f in filters), your_list )

这样做的好处是,只要单个过滤器返回False,就不会包含该列表元素.

Answer 3

生成器表达式似乎是最惯用的方法(并且您可以免费获得懒惰):

def nFilter(filters, tuples):
    return (t for t in tuples if all(f(t) for f in filters))

或等价物(可以说更具可读性):

def nFilter(filters, tuples):
    for tuple in tuples:
        if all(filter(tuple) for filter in filters):
            yield tuple

Answer 4

好吧,这里没有花哨的itertools等,只是使用一个简单的循环来避免递归和生成器的开销:

def for_loop(filters, tuples):
    for f in filters:
        tuples = filter(f, tuples)
        if not tuples: 
            return tuples
    return tuples

这是一个有点脏的基准:

import datetime
from itertools import ifilter
from timeit import Timer

def f1(t): return t[3]<10
def f2(t): return t[0]!=1
def f3(t): return t[1] in ("lisa","eric")
def f4(t): return t[3]>2

def original(filters,tuples):
    if filters and tuples:
        return original(filters,filter(filters.pop(),tuples))
    else: 
        return tuples

def filter_lambda_all(filters, tuples):
    return filter(lambda t: all(f(t) for f in filters), tuples)

def loop(filters, tuples):
    while filters and tuples:
        f = filters[0]
        del filters[0]
        tuples = filter(f, tuples)
    return tuples

def pop_loop(filters, tuples):
    while filters and tuples:
        tuples = filter(filters.pop(), tuples)
    return tuples

def for_loop(filters, tuples):
    for f in filters:
        tuples = filter(f, tuples)
        if not tuples: 
            return tuples
    return tuples


def with_ifilter(filters, tuples):
    for f in filters:
        tuples = ifilter(f, tuples)
    return tuples

_filters = [f1, f2, f3, f4]

def time(f):
    def t():
        return [    (0,'tom','...',8),
                    (1,'john','...',17),
                    (2,'lisa','...',1),
                    (3,'eric','...',18)
                ]*1000
    for i in xrange(4):
        list(f(_filters[i:] * 15,t()))

if __name__=='__main__':
    for f in (original,filter_lambda_all,loop,pop_loop,with_ifilter,for_loop):
        t = Timer(lambda: time(f))
        d = t.timeit(number=400)
        print f.__name__, d

结果:

原始7.23815271085
filter_lambda_all 14.1629812265
loop 7.23445844453
pop_loop 7.3084566637
with_ifilter 9.2767674205
for_loop 7.02854999945

Answer 5

我建议使用以下模式在发生器上自由应用一系列/过滤器链:

from functools import reduce, partial
from itertools import ifilter

filtered = reduce(lambda s,f: ifilter(f,s), filter_set, unfiltered)

简而言之,它在生成器上从左到右设置了一系列过滤器,并返回生成器,这是在原件上应用所有过滤器的结果.

如果您想获得一个列表,以下就足够了:

[reduce(lambda s,f: ifilter(f,s), (f1,f2,f3,), tuples)]

如果您希望获得单个功能,可以将其定义为:

chain_filters = partial(reduce, lambda s,f: ifilter(f,s))

并用作:

[chain_filters((f1,f2,f3,), tuples)]

请注意,此解决方案不会组成过滤器(如all()),而是将它们链接起来.如果您正在使用一些繁重的计算,您可能希望将更积极的过滤器放在链的开头,例如数据库查询过滤器之前的布隆过滤器等.