Redis Advanced Data Structures: Streams, HyperLogLog, and Probabilistic

Redis Is More Than a Cache

Most developers use Redis as a key-value store. But it has rich data structures that solve entire categories of problems.

Redis Streams (Kafka-lite)

import Redis from 'ioredis'
const redis = new Redis()

// Producer: add events to stream
await redis.xadd('events:user', '*',  // * = auto-generate ID
  'type', 'USER_CREATED',
  'userId', '123',
  'email', 'alice@example.com'
)

// Consumer: read from stream
const messages = await redis.xread(
  'COUNT', 10,
  'BLOCK', 5000,        // block 5s waiting for new messages
  'STREAMS', 'events:user',
  '
#39;                   // $ = only new messages
)

// Consumer Groups (for competing consumers)
await redis.xgroup('CREATE', 'events:user', 'processors', '
#39;, 'MKSTREAM')

// Read as consumer in group
const pending = await redis.xreadgroup(
  'GROUP', 'processors', 'worker-1',
  'COUNT', 10,
  'BLOCK', 5000,
  'STREAMS', 'events:user', '>'
)

// Acknowledge processed messages
await redis.xack('events:user', 'processors', messageId)

// Check pending (unacknowledged) messages
const info = await redis.xpending('events:user', 'processors', '-', '+', 10)

HyperLogLog (Unique Visitor Counting)

// Count unique page visitors without storing all IDs
// Uses only 12KB regardless of cardinality!

async function trackVisit(page: string, userId: string) {
  await redis.pfadd(`visitors:${page}:${getTodayKey()}`, userId)
}

async function getUniqueVisitors(page: string): Promise<number> {
  return redis.pfcount(`visitors:${page}:${getTodayKey()}`)
}

// Merge multiple counters
async function getWeeklyUniqueVisitors(page: string): Promise<number> {
  const keys = getLastNDayKeys(7).map(day => `visitors:${page}:${day}`)
  await redis.pfmerge(`visitors:${page}:week`, ...keys)
  return redis.pfcount(`visitors:${page}:week`)
}

Bloom Filters (with RedisBloom)

// Check membership without false negatives
// Small false positive rate (~1%)

// Has this email been used before?
await redis.bf.add('used:emails', 'alice@example.com')

const exists = await redis.bf.exists('used:emails', 'newuser@example.com')
if (exists) {
  // Might exist — do DB lookup to confirm
} else {
  // Definitely doesn't exist — skip DB lookup
}

// Create bloom filter with custom error rate
await redis.bf.reserve('bf:urls', 0.001, 1_000_000) // 0.1% error, 1M items

Sorted Sets for Rankings and Time Windows

// Real-time leaderboard
await redis.zadd('scores', 9500, 'player:alice')
await redis.zadd('scores', 8200, 'player:bob')
await redis.zadd('scores', 9800, 'player:charlie')

// Top 10 players
const top10 = await redis.zrevrange('scores', 0, 9, 'WITHSCORES')

// Player rank
const rank = await redis.zrevrank('scores', 'player:alice')  // 0-indexed

// Time-windowed sliding window counter (for rate limiting, analytics)
const now = Date.now()
const windowMs = 60 * 1000

await redis.zadd('api:calls:user:123', now, now.toString())
await redis.zremrangebyscore('api:calls:user:123', '-inf', now - windowMs)
const callsInWindow = await redis.zcard('api:calls:user:123')

Redis Time Series

// Store metrics efficiently
await redis.call('TS.CREATE', 'cpu:usage', 'RETENTION', 86400000, 'LABELS', 'host', 'server1')

// Add data point
await redis.call('TS.ADD', 'cpu:usage', '*', '75.5')

// Query range
const data = await redis.call('TS.RANGE', 'cpu:usage',
  Date.now() - 3600000,  // 1 hour ago
  '+',
  'AGGREGATION', 'avg', '60000'  // 1-minute averages
)

Geo Spatial Queries

// Add locations
await redis.geoadd('restaurants',
  -73.9857, 40.7484, 'restaurant:1',  // lng, lat, member
  -73.9901, 40.7580, 'restaurant:2'
)

// Find nearby restaurants (within 1km)
const nearby = await redis.georadius(
  'restaurants',
  -73.9857, 40.7484,
  1, 'km',
  'WITHCOORD', 'WITHDIST', 'ASC', 'COUNT', 10
)

Lua Scripts for Atomic Operations

const incrementIfLess = `
local current = redis.call('GET', KEYS[1])
if current == false then
  redis.call('SET', KEYS[1], ARGV[1])
  return tonumber(ARGV[1])
end
if tonumber(current) < tonumber(ARGV[2]) then
  redis.call('SET', KEYS[1], ARGV[1])
  return tonumber(ARGV[1])
end
return tonumber(current)
`

const result = await redis.eval(incrementIfLess, 1, 'counter', '1', '100')