Redis Advanced Data Structures in Production

Most developers use Redis as a cache or session store. But Redis includes sophisticated data structures that solve specific problems elegantly.

Redis Streams: Durable Event Log

Streams provide persistent, append-only messaging with consumer groups:

import redis
r = redis.Redis(decode_responses=True)

# Producer: append event to stream
msg_id = r.xadd(
    'user-events',
    { 'user_id': '12345', 'event_type': 'purchase', 'amount': '99.99' },
    maxlen=1000000, approximate=True
)

# Consumer Group: distributed processing
r.xgroup_create('user-events', 'payment-processors', id='
#39;, mkstream=True)

messages = r.xreadgroup(
    groupname='payment-processors',
    consumername='worker-1',
    streams={'user-events': '>'},  # Only undelivered messages
    count=10, block=2000
)

for stream, events in messages:
    for event_id, data in events:
        try:
            process_payment(data)
            r.xack('user-events', 'payment-processors', event_id)
        except Exception:
            pass  # Message stays in pending list for retry

# Reclaim stale messages from dead workers
r.xautoclaim('user-events', 'payment-processors', 'worker-2',
             min_idle_time=300000, start_id='0-0')

HyperLogLog: Count Billions Uniquely

Use only 12KB memory regardless of dataset size (0.81% error rate):

def track_page_view(page_id: str, user_id: str):
    key = f"hll:views:{page_id}:{datetime.date.today()}"
    r.pfadd(key, user_id)
    r.expire(key, 86400 * 30)

def get_unique_visitors(page_id: str) -> int:
    return r.pfcount(f"hll:views:{page_id}:{datetime.date.today()}")

def visitors_across_pages(page_ids: list) -> int:
    keys = [f"hll:views:{pid}:{datetime.date.today()}" for pid in page_ids]
    temp = f"hll:temp:{uuid.uuid4()}"
    r.pfmerge(temp, *keys)
    r.expire(temp, 60)
    return r.pfcount(temp)

# Memory comparison for 100M unique users:
# Python Set: ~3.2GB
# HyperLogLog: 12KB (267,000x less!)

Bloom Filters

r.bf().create('seen_emails', error_rate=0.01, capacity=10_000_000)

def process_email(email: str) -> bool:
    if r.bf().exists('seen_emails', email):
        return False  # Probably seen (1% false positive)
    r.bf().add('seen_emails', email)
    process_new_email(email)
    return True

# Cuckoo Filter supports deletion (Bloom filters don't)
r.cf().create('active_sessions', capacity=1_000_000)
r.cf().add('active_sessions', session_token)
r.cf().delete('active_sessions', session_token)  # On logout

Sorted Sets: Leaderboards and Rate Limiting

class Leaderboard:
    def __init__(self, game_id: str):
        self.key = f"leaderboard:{game_id}"

    def add_score(self, player_id: str, score: float):
        r.zadd(self.key, {player_id: score})

    def get_rank(self, player_id: str):
        rank = r.zrevrank(self.key, player_id)
        return rank + 1 if rank is not None else None

    def get_top(self, n: int):
        return r.zrevrange(self.key, 0, n-1, withscores=True)

    def get_nearby(self, player_id: str, window: int = 5):
        rank = r.zrevrank(self.key, player_id)
        if rank is None: return []
        return r.zrevrange(self.key, max(0, rank-window), rank+window, withscores=True)


class SlidingWindowRateLimiter:
    def is_allowed(self, user_id: str, max_req: int, window_sec: int) -> bool:
        key = f"rate:{user_id}"
        now = time.time()
        pipe = r.pipeline()
        pipe.zremrangebyscore(key, 0, now - window_sec)
        pipe.zcard(key)
        pipe.zadd(key, {str(now): now})
        pipe.expire(key, window_sec * 2)
        count = pipe.execute()[1]
        return count < max_req

Lua Scripting: Atomic Operations

# Atomic distributed lock using Lua script
acquire_script = r.register_script(
    "if redis.call('exists', KEYS[1]) == 0 then "
    "redis.call('set', KEYS[1], ARGV[1], 'px', tonumber(ARGV[2])) "
    "return 1 end return 0"
)

release_script = r.register_script(
    "if redis.call('get', KEYS[1]) == ARGV[1] then "
    "redis.call('del', KEYS[1]) return 1 end return 0"
)

class RedisLock:
    def __init__(self, resource: str, ttl_ms: int = 30000):
        self.key = f"lock:{resource}"
        self.token = str(uuid.uuid4())
        self.ttl_ms = ttl_ms

    def acquire(self, timeout: float = 10) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if acquire_script(keys=[self.key], args=[self.token, self.ttl_ms]):
                return True
            time.sleep(0.05)
        return False

    def release(self):
        release_script(keys=[self.key], args=[self.token])

    def __enter__(self):
        if not self.acquire():
            raise TimeoutError(f"Cannot acquire lock: {self.key}")
        return self

    def __exit__(self, *_):
        self.release()

with RedisLock("payment:12345"):
    process_payment(12345)

Pub/Sub vs Streams

Pub/Sub: Fire-and-forget broadcasting. Messages lost if offline. Use for real-time notifications.

Streams: Persistent, replayable log. Consumer groups for distributed processing. Use for event sourcing, task queues, audit logs.

Redis's specialized data structures eliminate entire services: Sorted Set leaderboards outperform dedicated services, HyperLogLog handles unique counting at any scale for free, and Streams replace simple message queues.