python - 如何将python集与redis集相交

Question

我有一组 python 代码，比方说 item_ids

item_ids = {1, 2, 3, 4, 5}

而且我还有一个预先计算好的 redis 集，items_with_property

127.0.0.1:6379> SADD items_with_property 3 4 5 6
(integer) 5
127.0.0.1:6379> SMEMBERS items_with_property
1) "3"
2) "4"
3) "5"
4) "6"

是否有可能在不将 items_with_property 提取到 python 内存的情况下将 item_ids 与 items_with_property 相交？

我想要类似的东西

>>> print(redis_client.magic_sinter(item_ids, 'items_with_property'))
{3, 4, 5}

这样做的原因是 items_with_property 和 item_ids 可以非常大，我不想在服务器之间传输大量数据(redis 是在单独的机器上，并且有很多客户)

更新 2019-02-05

我为不同的方法准备了一个速度测试，这是我的结果:

from django.core.cache import caches
from django.utils.functional import cached_property
from django_redis import get_redis_connection
from hot_redis import Set

search_storage = caches['search.filter']


class Command(BaseCommand):

    command_name = 'performance_test'

    def handle(self, *args, **options):
        """
        output:
        1
            TEST1 results time:  13.432172060012817
            TEST2 results time:  4.478500127792358
            TEST3 results time:  4.45565390586853
            TEST4 results time:  4.674767732620239
            TEST5 results time:  3.244804859161377
            TEST6 results time:  4.4963860511779785

        2
            TEST1 results time:  13.012064695358276
            TEST2 results time:  4.4086668491363525
            TEST3 results time:  4.4962310791015625
            TEST4 results time:  4.745664119720459
            TEST5 results time:  3.3029701709747314
            TEST6 results time:  4.676959991455078

        3
            TEST1 results time:  12.83815312385559
            TEST2 results time:  4.190127849578857
            TEST3 results time:  4.445873260498047
            TEST4 results time:  4.724813938140869
            TEST5 results time:  3.2511937618255615
            TEST6 results time:  4.454891920089722
        4
            TEST1 results time:  13.131163358688354
            TEST2 results time:  4.265545129776001
            TEST3 results time:  4.440964221954346
            TEST4 results time:  4.571079969406128
            TEST5 results time:  3.279599189758301
            TEST6 results time:  4.366865873336792
        5
            TEST1 results time:  13.424093961715698
            TEST2 results time:  4.349413156509399
            TEST3 results time:  4.42648720741272
            TEST4 results time:  4.607520818710327
            TEST5 results time:  3.415123224258423
            TEST6 results time:  4.391672134399414
        """
        item_ids = set(random.sample(range(10000000, 1000000000), 100000))  # 100k random ints

        # TEST1 - PYTHON INTERSECTION
        _started_at = time.time()
        _key = f'test1'
        search_storage.set(_key, item_ids)  # python pickled set
        for _ in range(1000):
            search_ids = set(random.sample(item_ids, k=100))
            redis_ids = search_storage.get(_key)
            result = search_ids & redis_ids
            assert len(result) == 100
        search_storage.delete(_key)
        print("TEST1 results time: ", time.time() - _started_at)

        # TEST2 - REDIS INTERSECTION, using stored function and SISMEMBER for every search_id
        _started_at = time.time()
        _key = f'test2'
        redis_con = get_redis_connection('search.filter')  # raw connetction for redis methods.
        redis_con.sadd(_key, *item_ids)
        stored_func = redis_con.register_script('''
        local reply = {}
        while #ARGV > 0 do
          local member = table.remove(ARGV)
          if redis.call('SISMEMBER', KEYS[1], member) == 1 then
            table.insert(reply, member)
          end
        end
        return reply
        ''')
        for _ in range(1000):
            search_ids = random.sample(item_ids, k=100)
            result = stored_func(keys=[_key], args=search_ids)
            assert len(result) == 100
        redis_con.delete(_key)
        print("TEST2 results time: ", time.time() - _started_at)

        # TEST3 - REDIS INTERSECTION, using python-made temp key
        _started_at = time.time()
        _key = f'test3'
        redis_con = get_redis_connection('search.filter')
        redis_con.sadd(_key, *item_ids)
        for _ in range(1000):
            search_ids = frozenset(random.sample(item_ids, k=100))
            _temp_key = f'test3_temp_{hash(search_ids)}'
            redis_con.sadd(_temp_key, *search_ids)
            result = redis_con.sinter(keys=[_key, _temp_key])
            redis_con.delete(_temp_key)
            assert len(result) == 100
        redis_con.delete(_key)
        print("TEST3 results time: ", time.time() - _started_at)

        # TEST4 - REDIS INTERSECTION, using stored function and redis-made temp key
        _started_at = time.time()
        _key = f'test4'
        redis_con = get_redis_connection('search.filter')
        redis_con.sadd(_key, *item_ids)
        stored_func = redis_con.register_script('''
        local reply = {}
        local temp_key = KEYS[1]
        redis.call('SADD', temp_key, unpack(ARGV))
        reply = redis.call('SINTER', temp_key, KEYS[2])
        redis.call('DEL', temp_key)
        return reply
        ''')
        for _ in range(1000):
            search_ids = frozenset(random.sample(item_ids, k=100))
            _temp_key = f'test4_temp_{hash(search_ids)}'
            result = stored_func(keys=[_temp_key, _key], args=search_ids)
            assert len(result) == 100
        redis_con.delete(_key)
        print("TEST4 results time: ", time.time() - _started_at)

        # TEST5 - PYTHON INTERSECTION, using cached_property
        _started_at = time.time()
        _key = f'test5'
        search_storage.set(_key, item_ids, SEARCH_FILTER_TIMEOUT)
        for _ in range(1000):
            search_ids = set(random.sample(item_ids, k=100))
            redis_ids = self.cached_cached_item_ids
            result = search_ids & redis_ids
            assert len(result) == 100
        search_storage.delete(_key)
        print("TEST5 results time: ", time.time() - _started_at)

        # TEST6 - HOT REDIS
        _started_at = time.time()
        _key = f'test6'
        hot_items = Set(key=_key, initial=item_ids)
        for _ in range(1000):
            search_ids = Set(initial=random.sample(item_ids, k=100))
            result = hot_items & search_ids
            search_ids.clear()
            assert len(result) == 100
        hot_items.clear()
        print("TEST6 results time: ", time.time() - _started_at)

    @cached_property
    def cached_cached_item_ids(self):
        return search_storage.get('test5')

你可以自己试试，这是一个Django命令，但我认为重点很明确。

获胜者是 TEST5 - “本地缓存”缓存结果。我们大大减少了序列化数据的次数，但我们还有另一个头疼的问题 - 如何使第二层缓存失效。

对我来说，赢家是 hot-redis - python 库已经实现了所有合适的 lua methods ，并且有干净的界面。

Answer 1

使用 Lua(另一种语言)在任何语言中都有一种神奇的方法可以做到这一点。参见 Redis 的 EVAL和 redis-py 的 Script 辅助类。

脚本需要 Redis 集的键名和代表客户端集的任意数量的附加参数('item_ids')。对于每个参数，它对目标集执行 SISMEMBER 操作以确定相交。

from redis import Redis

items_with_property = [3, 4, 5, 6]
item_ids = [1, 2, 3, 4, 5]

r = Redis()
r.sadd('items_with_property', *items_with_property)
s = r.register_script('''
local reply = {}
while #ARGV > 0 do
  local member = table.remove(ARGV)
  if redis.call('SISMEMBER', KEYS[1], member) == 1 then
    table.insert(reply, member)
  end
end
return reply
''')
print(s(keys=['items_with_property'], args=item_ids))

注意:另一种方法是使用临时 key 来存储用户提供的成员，然后对源集和临时集执行 SINTER 操作。对于较大的本地 Set，这应该会表现更好。