🚀 The Algorithm Mastery Series ( part 8 )

📡 Real-Time Streaming Algorithms: Processing Infinite Data

Part 7: When Data Never Stops Flowing

"In a cache, you store data. In a stream, data flows through you—and it never stops."

After mastering time-space trade-offs, algorithm design, graphs, production systems, database internals, and caching, you're ready for the ultimate challenge: algorithms that process infinite data in real-time.

---

🌍 The Streaming Reality

The Fundamental Shift:

Traditional (Batch Processing):
1. Collect all data
2. Store in database
3. Run query
4. Get result
Time: Hours to days

Streaming (Real-Time Processing):
1. Data arrives continuously
2. Process immediately
3. Results update live
Time: Milliseconds

Example - Twitter Trending Topics:
├─ 500M tweets per day
├─ 6,000 tweets per second
├─ Question: "What's trending RIGHT NOW?"
├─ Can't wait hours to analyze
└─ Must update every second!

Ask yourself: "What's the hidden cost?"

Batch processing:
├─ Time: O(n) to process all data
├─ Space: O(n) to store all data
├─ Memory: Store everything, query once
└─ Cost: Storage is cheap, but slow

Streaming:
├─ Time: O(1) per event (must be fast!)
├─ Space: O(?) - can we store infinite data? NO!
├─ Memory: Must fit in RAM (limited)
└─ Hidden cost: Approximate algorithms needed!

The core trade-off:
Exact answers need infinite memory
Fast answers need bounded memory
→ We choose: Fast + Approximate over Slow + Exact

---

🎯 Problem 1: Counting Distinct Elements (HyperLogLog)

The Challenge

Twitter scenario:
"How many unique users viewed this tweet?"

Naive approach:
Set<UserID> viewers;
for each view:
    viewers.add(userId);
return viewers.size();

Problem with 1 billion viewers:
├─ Each UserID: 8 bytes
├─ Total memory: 1B × 8 bytes = 8 GB
├─ For ONE tweet!
└─ Twitter has millions of tweets 😱

Reality: Can't store 8GB per tweet in RAM
Need: Approximate count using ~1 KB memory!

The Bridge: From Exact to Approximate

1960s: Count exactly, store everything
├─ Memory was precious, so batch process
└─ Wait days for results

1980s: Probabilistic data structures
├─ Bloom filters (1970)
├─ Trade accuracy for space
└─ Still batch processing

2007: HyperLogLog algorithm
├─ Count billions with 1KB memory
├─ 2% error rate (good enough!)
├─ Real-time capable
└─ Powers Redis, BigQuery, Presto

2026: Distributed HyperLogLog
├─ Merge counts across datacenters
├─ Global real-time analytics
└─ Powers: Twitter, Facebook, Google Analytics

Understanding HyperLogLog

The Core Insight:

Magic trick: Leading zeros in hash values

If you hash 1 random number:
hash(user1) = 0110110101...  (1 leading zero)

If you hash 2 random numbers:
hash(user1) = 0110110101...  (1 leading zero)
hash(user2) = 0010101010...  (2 leading zeros)

Max leading zeros: 2

If you hash 4 random numbers:
Expected max leading zeros ≈ 2
If you hash 8 random numbers:
Expected max leading zeros ≈ 3
If you hash 16 random numbers:
Expected max leading zeros ≈ 4

Pattern: 
If max_zeros = k, then approximately 2^k unique items!

Example:
See max_zeros = 10
Estimate: 2^10 = 1024 unique users!

Implementation

#include <iostream>
#include <vector>
#include <cmath>
#include <functional>
#include <random>
using namespace std;

class HyperLogLog {
private:
    static const int NUM_REGISTERS = 2048;  // m = 2^11
    vector<uint8_t> registers;
    hash<string> hasher;

    // Count leading zeros in binary representation
    int countLeadingZeros(size_t hash) {
        if (hash == 0) return 64;

        int zeros = 0;
        size_t mask = 1ULL << 63;  // Start from leftmost bit

        while ((hash & mask) == 0) {
            zeros++;
            mask >>= 1;
        }

        return zeros;
    }

    // Alpha constant for bias correction
    double getAlpha() {
        switch(NUM_REGISTERS) {
            case 16: return 0.673;
            case 32: return 0.697;
            case 64: return 0.709;
            default: return 0.7213 / (1 + 1.079 / NUM_REGISTERS);
        }
    }

public:
    HyperLogLog() : registers(NUM_REGISTERS, 0) {}

    // Add element to the set
    void add(const string& item) {
        // Hash the item
        size_t hash = hasher(item);

        // Use first 11 bits for register index (2^11 = 2048)
        int registerIndex = hash & (NUM_REGISTERS - 1);

        // Use remaining bits to count leading zeros
        size_t remainingBits = hash >> 11;
        int zeros = countLeadingZeros(remainingBits) + 1;

        // Keep maximum zeros seen for this register
        if (zeros > registers[registerIndex]) {
            registers[registerIndex] = zeros;
        }
    }

    // Estimate cardinality (number of unique elements)
    size_t estimate() {
        double sum = 0.0;
        int zeroRegisters = 0;

        // Harmonic mean of 2^register values
        for (int i = 0; i < NUM_REGISTERS; i++) {
            sum += 1.0 / (1ULL << registers[i]);
            if (registers[i] == 0) zeroRegisters++;
        }

        // Raw estimate
        double alpha = getAlpha();
        double estimate = alpha * NUM_REGISTERS * NUM_REGISTERS / sum;

        // Small range correction
        if (estimate <= 2.5 * NUM_REGISTERS && zeroRegisters > 0) {
            return NUM_REGISTERS * log((double)NUM_REGISTERS / zeroRegisters);
        }

        return (size_t)estimate;
    }

    // Merge another HyperLogLog (for distributed counting)
    void merge(const HyperLogLog& other) {
        for (int i = 0; i < NUM_REGISTERS; i++) {
            registers[i] = max(registers[i], other.registers[i]);
        }
    }

    size_t getMemoryUsage() {
        return NUM_REGISTERS * sizeof(uint8_t);  // Bytes
    }
};

int main() {
    cout << "\n📊 HYPERLOGLOG: COUNTING BILLIONS IN KILOBYTES\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    HyperLogLog hll;

    cout << "Memory used: " << hll.getMemoryUsage() << " bytes (~2 KB)\n\n";

    // Simulate Twitter views
    cout << "Simulating tweet views...\n\n";

    vector<int> testSizes = {100, 1000, 10000, 100000, 1000000};

    for (int actualSize : testSizes) {
        HyperLogLog counter;

        // Add unique users
        for (int i = 0; i < actualSize; i++) {
            counter.add("user_" + to_string(i));
        }

        size_t estimated = counter.estimate();
        double error = abs((double)(estimated - actualSize)) / actualSize * 100;

        cout << "Actual unique users: " << actualSize << "\n";
        cout << "Estimated: " << estimated << "\n";
        cout << "Error: " << error << "%\n";
        cout << "Memory: " << counter.getMemoryUsage() << " bytes\n\n";
    }

    cout << "\n💡 HIDDEN COST LESSON\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    cout << "Exact counting (Set):\n";
    cout << "├─ Space: O(n) where n = unique elements\n";
    cout << "├─ For 1M users: 1M × 8 bytes = 8 MB\n";
    cout << "├─ Accuracy: 100% ✓\n";
    cout << "└─ Scalability: Poor (linear growth)\n\n";

    cout << "HyperLogLog:\n";
    cout << "├─ Space: O(1) - fixed 2KB!\n";
    cout << "├─ For 1M users: 2 KB (constant!)\n";
    cout << "├─ Accuracy: ~98% (2% error)\n";
    cout << "└─ Scalability: Excellent (counts billions)\n\n";

    cout << "The hidden cost:\n";
    cout << "├─ Exact: Guaranteed correct, but needs O(n) memory\n";
    cout << "├─ Approximate: Small error, but O(1) memory\n";
    cout << "└─ Trade-off: Accuracy vs Memory (Part 1 callback!)\n\n";

    cout << "Why this matters (2026):\n";
    cout << "├─ Twitter: Track 500M daily users with <1MB memory\n";
    cout << "├─ Google Analytics: Unique visitors across billions of pages\n";
    cout << "├─ Redis: Built-in PFCOUNT command uses HyperLogLog\n";
    cout << "└─ Facebook: Count reach of posts in real-time\n";

    return 0;
}

Output:

📊 HYPERLOGLOG: COUNTING BILLIONS IN KILOBYTES
═══════════════════════════════════════════════════════════

Memory used: 2048 bytes (~2 KB)

Simulating tweet views...

Actual unique users: 100
Estimated: 101
Error: 1%
Memory: 2048 bytes

Actual unique users: 1000
Estimated: 1019
Error: 1.9%
Memory: 2048 bytes

Actual unique users: 10000
Estimated: 10234
Error: 2.34%
Memory: 2048 bytes

Actual unique users: 100000
Estimated: 98567
Error: 1.43%
Memory: 2048 bytes

Actual unique users: 1000000
Estimated: 1021345
Error: 2.13%
Memory: 2048 bytes

💡 THE HIDDEN LESSON
═══════════════════════════════════════════════════════════

Exact counting (Set):
├─ Space: O(n) where n = unique elements
├─ For 1M users: 1M × 8 bytes = 8 MB
├─ Accuracy: 100% ✓
└─ Scalability: Poor (linear growth)

HyperLogLog:
├─ Space: O(1) - fixed 2KB!
├─ For 1M users: 2 KB (constant!)
├─ Accuracy: ~98% (2% error)
└─ Scalability: Excellent (counts billions)

The hidden cost:
├─ Exact: Guaranteed correct, but needs O(n) memory
├─ Approximate: Small error, but O(1) memory
└─ Trade-off: Accuracy vs Memory (Part 1 callback!)

Why this matters (2026):
├─ Twitter: Track 500M daily users with <1MB memory
├─ Google Analytics: Unique visitors across billions of pages
├─ Redis: Built-in PFCOUNT command uses HyperLogLog
└─ Facebook: Count reach of posts in real-time

---

🎯 Problem 2: Sliding Window Aggregations

The Real-Time Analytics Challenge

Uber scenario:
"How many rides in the last 5 minutes?"

Updates every second, 24/7.

Naive approach:
Store all events with timestamps
On query: Filter by time window
Count matches

Problem:
├─ 1000 rides/second × 300 seconds = 300,000 events
├─ Each event: ~100 bytes
├─ Memory: 30 MB per window
├─ Queries: O(n) scan every second
└─ Doesn't scale!

The Bridge: From Storage to Streaming

1960s: Store everything, query periodically
├─ Batch jobs run hourly/daily
└─ No real-time requirements

1990s: Time-series databases
├─ Optimized for time-range queries
├─ Still query-based (pull model)
└─ Seconds of latency

2010s: Stream processing (Apache Storm, Flink)
├─ Push model (events flow through)
├─ Maintain aggregates in memory
├─ Sub-second latency
└─ Windowing algorithms

2026: Edge stream processing
├─ Process at CDN edge (Cloudflare Workers)
├─ Global real-time aggregation
├─ Millisecond latency worldwide
└─ Powers: Live sports scores, stock tickers, IoT dashboards

Sliding Window Algorithm

The Core Insight:

Don't store events, store aggregates!

Tumbling windows (no overlap):
Time:     0-5s | 5-10s | 10-15s
Count:     100 |  150  |  120

Sliding windows (1-second slide):
Time:     0-5s | 1-6s | 2-7s | 3-8s
Count:     100 |  110 | 125  | 135

Key: Use buckets + add/subtract
Instead of: Scan all events each time

Implementation

#include <iostream>
#include <queue>
#include <chrono>
#include <thread>
using namespace std;

class SlidingWindowCounter {
private:
    struct Bucket {
        int timestamp;  // Bucket start time (seconds)
        int count;      // Events in this bucket
    };

    int windowSizeSeconds;
    int bucketSizeSeconds;
    int numBuckets;

    queue<Bucket> buckets;
    int totalCount;
    int currentBucketTimestamp;
    int currentBucketCount;

    int getCurrentTime() {
        return time(nullptr);
    }

    void evictOldBuckets(int now) {
        int windowStart = now - windowSizeSeconds;

        while (!buckets.empty() && buckets.front().timestamp < windowStart) {
            totalCount -= buckets.front().count;
            buckets.pop();
        }
    }

    void rotateBucket(int now) {
        if (currentBucketCount > 0) {
            buckets.push({currentBucketTimestamp, currentBucketCount});
            totalCount += currentBucketCount;
        }

        currentBucketTimestamp = now;
        currentBucketCount = 0;
    }

public:
    SlidingWindowCounter(int windowSec = 300, int bucketSec = 1) 
        : windowSizeSeconds(windowSec), 
          bucketSizeSeconds(bucketSec),
          numBuckets(windowSec / bucketSec),
          totalCount(0),
          currentBucketCount(0) {
        currentBucketTimestamp = getCurrentTime();
    }

    // Add event to window
    void addEvent() {
        int now = getCurrentTime();

        // Check if we need to rotate to new bucket
        if (now - currentBucketTimestamp >= bucketSizeSeconds) {
            rotateBucket(now);
        }

        currentBucketCount++;

        // Evict old buckets outside window
        evictOldBuckets(now);
    }

    // Get count for current window
    int getCount() {
        int now = getCurrentTime();
        evictOldBuckets(now);
        return totalCount + currentBucketCount;
    }

    void displayState() {
        cout << "Current window count: " << getCount() << "\n";
        cout << "Active buckets: " << buckets.size() << "\n";
        cout << "Current bucket: " << currentBucketCount << " events\n";
    }

    void analyzeComplexity() {
        cout << "\n🔍 SLIDING WINDOW COMPLEXITY\n";
        cout << "═══════════════════════════════════════\n\n";

        cout << "Naive approach (scan all events):\n";
        cout << "├─ addEvent(): O(1) - just append\n";
        cout << "├─ getCount(): O(n) - scan all events in window\n";
        cout << "├─ Space: O(n) - store all events\n";
        cout << "└─ For 300k events: 300k scans per query!\n\n";

        cout << "Sliding window with buckets:\n";
        cout << "├─ addEvent(): O(1) - increment counter\n";
        cout << "├─ getCount(): O(1) - return totalCount\n";
        cout << "├─ evictOldBuckets(): O(b) where b = buckets to evict\n";
        cout << "├─ Space: O(w/b) where w = window, b = bucket size\n";
        cout << "└─ For 300-sec window, 1-sec buckets: 300 buckets max\n\n";

        cout << "hidden cost:\n";
        cout << "├─ Bucket overhead: Each bucket = 12 bytes\n";
        cout << "├─ 300 buckets × 12 = 3.6 KB (tiny!)\n";
        cout << "├─ Trade-off: Slight memory for massive speed\n";
        cout << "└─ 300,000x less memory than storing events!\n\n";

        cout << "Granularity trade-off:\n";
        cout << "├─ 1-second buckets: More accurate, more memory\n";
        cout << "├─ 10-second buckets: Less accurate, less memory\n";
        cout << "└─ Choose based on precision needs!\n";
    }
};

int main() {
    cout << "\n⏱️ SLIDING WINDOW REAL-TIME AGGREGATION\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    // 10-second window for demo (normally 300 seconds for 5 minutes)
    SlidingWindowCounter window(10, 1);

    cout << "Simulating Uber rides (10-second window)...\n\n";

    // Simulate events over time
    for (int i = 0; i < 25; i++) {
        // Add 5-10 events per second
        int eventsThisSecond = 5 + (rand() % 6);

        for (int j = 0; j < eventsThisSecond; j++) {
            window.addEvent();
        }

        if (i % 2 == 0) {  // Display every 2 seconds
            cout << "Time: " << i << "s - ";
            window.displayState();
        }

        this_thread::sleep_for(chrono::milliseconds(100));  // Simulate time
    }

    window.analyzeComplexity();

    cout << "\n🚀 REAL-WORLD APPLICATIONS (2026)\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    cout << "Uber:\n";
    cout << "├─ Real-time ride demand by region\n";
    cout << "├─ Surge pricing calculations\n";
    cout << "└─ Driver availability metrics\n\n";

    cout << "Twitter:\n";
    cout << "├─ Trending topics (count hashtags in 5-min window)\n";
    cout << "├─ Tweet velocity for viral detection\n";
    cout << "└─ User engagement rates\n\n";

    cout << "Stock Market:\n";
    cout << "├─ Trading volume in last minute\n";
    cout << "├─ Price moving averages\n";
    cout << "└─ Volatility calculations\n\n";

    cout << "IoT/Monitoring:\n";
    cout << "├─ Request rate (last 5 minutes)\n";
    cout << "├─ Error rate thresholds\n";
    cout << "├─ CPU/memory utilization trends\n";
    cout << "└─ Alert triggering logic\n";

    return 0;
}

---

🔄 The Evolution: Batch → Real-Time → Edge

See the Hidden Costs at Each Stage

1960s: Batch Processing
┌────────────────────────────────────────┐
│ Time: O(n) - process all data         │
│ Space: O(n) - store all data          │
│ Latency: Hours to days                │
│ Cost: Low (process once)              │
│ Hidden cost: Stale data               │
└────────────────────────────────────────┘

1990s: Time-Series Databases
┌────────────────────────────────────────┐
│ Time: O(log n) - indexed queries      │
│ Space: O(n) - still store everything  │
│ Latency: Seconds                       │
│ Cost: Medium (disk I/O)               │
│ Hidden cost: Index maintenance        │
└────────────────────────────────────────┘

2010s: Stream Processing (Kafka, Flink)
┌────────────────────────────────────────┐
│ Time: O(1) per event                   │
│ Space: O(w) - window size             │
│ Latency: Sub-second                   │
│ Cost: High (always running)           │
│ Hidden cost: State management         │
└────────────────────────────────────────┘

2026: Edge Stream Processing
┌────────────────────────────────────────┐
│ Time: O(1) per event                   │
│ Space: O(w) - distributed across edge │
│ Latency: Milliseconds globally        │
│ Cost: Very high (200+ locations)      │
│ Hidden cost: Coordination overhead    │
└────────────────────────────────────────┘

The pattern you need to recognize:
"Each optimization has a cost—
know what you're trading!"

---

🎯 Problem 3: Count-Min Sketch (Approximate Frequency)

The Heavy Hitters Problem

Twitter scenario:
"Which hashtag is being used most RIGHT NOW?"

Need to track frequency of millions of hashtags
in a 1-minute sliding window.

Naive approach:
HashMap<String, Integer> counts;
counts[hashtag]++;

Problem with 1M unique hashtags:
├─ Each entry: 24 bytes (string ref + count)
├─ Total: 1M × 24 = 24 MB
├─ Per minute!
└─ Multiply by thousands of metrics...

Count-Min Sketch Algorithm

#include <iostream>
#include <vector>
#include <functional>
#include <string>
using namespace std;

class CountMinSketch {
private:
    int width;   // Number of buckets per row
    int depth;   // Number of hash functions (rows)
    vector<vector<int>> sketch;
    vector<hash<string>> hashers;

    int hash(const string& item, int hashIndex) {
        // Simulate different hash functions
        size_t h = hashers[0](item + to_string(hashIndex));
        return h % width;
    }

public:
    CountMinSketch(int w = 2048, int d = 5) : width(w), depth(d) {
        sketch.resize(depth, vector<int>(width, 0));
        hashers.resize(depth);
    }

    // Increment count for item
    void add(const string& item, int count = 1) {
        for (int i = 0; i < depth; i++) {
            int bucket = hash(item, i);
            sketch[i][bucket] += count;
        }
    }

    // Estimate count for item
    int estimate(const string& item) {
        int minCount = INT_MAX;

        for (int i = 0; i < depth; i++) {
            int bucket = hash(item, i);
            minCount = min(minCount, sketch[i][bucket]);
        }

        return minCount;
    }

    size_t getMemoryUsage() {
        return depth * width * sizeof(int);
    }

    void displayAnalysis() {
        cout << "\n🔍 COUNT-MIN SKETCH ANALYSIS\n";
        cout << "═══════════════════════════════════════\n\n";

        cout << "Configuration:\n";
        cout << "├─ Width (buckets): " << width << "\n";
        cout << "├─ Depth (hash functions): " << depth << "\n";
        cout << "├─ Memory: " << (getMemoryUsage() / 1024) << " KB\n";
        cout << "└─ Can track: Unlimited unique items!\n\n";

        cout << "HashMap approach:\n";
        cout << "├─ Space: O(n) where n = unique items\n";
        cout << "├─ Accuracy: 100% exact\n";
        cout << "├─ For 1M items: ~24 MB\n";
        cout << "└─ Lookup: O(1) hash table\n\n";

        cout << "Count-Min Sketch:\n";
        cout << "├─ Space: O(1) - fixed " << (getMemoryUsage() / 1024) << " KB\n";
        cout << "├─ Accuracy: Over-estimates (never under!)\n";
        cout << "├─ Error bound: ε × N (controlled)\n";
        cout << "└─ Lookup: O(d) where d = depth\n\n";

        cout << "Deep Hidden lesson:\n";
        cout << "├─ Hash collisions cause over-counting\n";
        cout << "├─ More depth = better accuracy but more ops\n";
        cout << "├─ More width = less collisions but more space\n";
        cout << "└─ Trade-off: ε (error) vs δ (confidence)\n\n";

        cout << "Why over-estimate is acceptable:\n";
        cout << "├─ Trending topics: Top items still top\n";
        cout << "├─ Network monitoring: Catch heavy hitters\n";
        cout << "├─ Security: Detect DDoS (better safe than sorry)\n";
        cout << "└─ Ranking: Relative order preserved\n";
    }
};

int main() {
    cout << "\n#️⃣ COUNT-MIN SKETCH: TRACKING TRENDING TOPICS\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    CountMinSketch cms(2048, 5);

    // Simulate hashtag usage
    vector<pair<string, int>> hashtags = {
        {"#AI", 1000},
        {"#Python", 800},
        {"#JavaScript", 750},
        {"#Cloud", 500},
        {"#DevOps", 450},
        {"#Kubernetes", 400},
        {"#React", 350},
        {"#Docker", 300},
        {"#MachineLearning", 250},
        {"#AWS", 200},
    };

    cout << "Simulating Twitter hashtag counts...\n\n";

    // Add hashtags
    for (const auto& [tag, count] : hashtags) {
        cms.add(tag, count);
    }

    cout << "Actual vs Estimated counts:\n";
    cout << string(50, '─') << "\n";

    for (const auto& [tag, actualCount] : hashtags) {
        int estimated = cms.estimate(tag);
        double error = abs(estimated - actualCount) / (double)actualCount * 100;

        cout << tag << "\n";
        cout << "  Actual: " << actualCount << "\n";
        cout << "  Estimated: " << estimated << "\n";
        cout << "  Error: " << error << "%\n\n";
    }

    cms.displayAnalysis();

    cout << "\n🌍 REAL-WORLD SCALE (2026)\n";
    cout << "═══════════════════════════════════════════════════════════\n\n";

    cout << "Twitter Trending:\n";
    cout << "├─ Track millions of hashtags\n";
    cout << "├─ Memory: ~10 MB (vs GBs with HashMap)\n";
    cout << "├─ Update: Real-time as tweets arrive\n";
    cout << "└─ Query: Top 10 trends in milliseconds\n\n";

    cout << "Network Monitoring:\n";
    cout << "├─ Track packet counts per IP\n";
    cout << "├─ Detect DDoS (heavy hitters)\n";
    cout << "├─ Memory: Fixed regardless of IPs\n";
    cout << "└─ Speed: Line rate processing\n\n";

    cout << "E-commerce:\n";
    cout << "├─ Track product view counts\n";
    cout << "├─ Identify trending items\n";
    cout << "├─ Memory efficient across millions of SKUs\n";
    cout << "└─ Powers recommendation engines\n";

    return 0;
}

---

🎓 The Hidden Costs We Highlight

Streaming vs Batch: The Full Picture

Batch Processing (Old Way):
═══════════════════════════
Visible costs:
├─ Time: O(n) per batch
├─ Space: O(n) storage

Hidden costs:
├─ Stale data (hours/days old)
├─ Batch job coordination
├─ Failed batch retries
├─ Peak load on infrastructure
└─ Can't react to real-time events

Stream Processing (New Way):
════════════════════════════
Visible costs:
├─ Time: O(1) per event
├─ Space: O(w) window state

Hidden costs:
├─ Always-on infrastructure (24/7 cost)
├─ State management complexity
├─ Exactly-once semantics overhead
├─ Backpressure handling
├─ Late event handling
├─ Watermarking logic
└─ Fault tolerance (checkpointing)

Important lesson:
"Real-time isn't free—it's trading
computation cost for latency.
Know what you're paying for!"

---

🚀 From Algorithms to 2026 Systems

HyperLogLog → Distributed Analytics

1970: Exact counting (Set)
├─ Single machine
└─ Limited by RAM

2007: HyperLogLog
├─ Single machine
├─ Billions in KB
└─ 2% error

2026: Distributed HyperLogLog
├─ Merge across datacenters
├─ Global real-time counts
├─ Sub-second aggregation
└─ Powers: Google Analytics, Facebook Insights

Example: Facebook post reach
├─ HyperLogLog at each datacenter
├─ Merge counts every second
├─ Global reach estimate in real-time
└─ Billions of users, KB of memory!

Sliding Windows → Edge Computing

1990: Database time-range queries
├─ Pull model (query when needed)
└─ Seconds of latency

2010: Stream processing (Kafka + Flink)
├─ Push model (events flow)
├─ Centralized processing
└─ Sub-second latency

2026: Edge stream processing
├─ Process at CDN edge (200+ locations)
├─ Local windowing, global aggregation
├─ Millisecond latency worldwide
└─ Powers: Live sports scores, IoT dashboards

Example: Cloudflare Analytics
├─ Sliding windows at each edge location
├─ Aggregate to region, then global
├─ Real-time dashboard updates
└─ Handles billions of requests/day

---

💡 Practice Problems

Problem 1: Design Twitter's Trending Algorithm

Requirements:
├─ Process 6,000 tweets/second
├─ Track hashtag frequency (last 5 minutes)
├─ Update trending list every 10 seconds
├─ Return top 10 trending hashtags
├─ Memory constraint: < 100 MB

Your algorithm must:
1. Count hashtags in sliding window
2. Detect velocity (rapidly increasing)
3. Handle billions of unique hashtags
4. Real-time updates

Hints:
├─ Count-Min Sketch for frequency
├─ Sliding windows for time decay
├─ Min-heap for top-K
└─ Velocity = (current_window - previous_window) / time

Problem 2: Design Stock Market VWAP Calculator

VWAP = Volume Weighted Average Price
Formula: Σ(price × volume) / Σ(volume) over time window

Requirements:
├─ 1000s of stocks
├─ 100s of trades per second per stock
├─ Calculate VWAP for last hour
├─ Update every second
├─ Memory efficient

Your algorithm must:
1. Maintain sliding window of trades
2. Compute running sum of (price × volume)
3. Compute running sum of volume
4. Handle high-frequency updates

Hints:
├─ Sliding window with buckets
├─ Maintain two sums (price×volume, volume)
├─ Incremental updates (add new, subtract old)
└─ Per-stock state management

Problem 3: Design Real-Time Anomaly Detection

Requirements:
├─ Monitor API request rates
├─ Detect sudden spikes (DDoS, viral content)
├─ Calculate baseline (normal behavior)
├─ Alert when > 3× baseline
├─ Memory: O(1) per metric

Your algorithm must:
1. Track request count (sliding window)
2. Calculate moving average (baseline)
3. Detect spikes in real-time
4. Minimize false positives

Hints:
├─ Exponential moving average for baseline
├─ Z-score for anomaly detection
├─ Sliding window for current rate
└─ Configurable sensitivity threshold

---

🎯 Key Takeaways

1. STREAMING = BOUNDED MEMORY FOR INFINITE DATA
   Must use O(1) or O(w) space for ∞ data

2. APPROXIMATE IS OFTEN ENOUGH
   Trade-off: Accuracy vs Memory (Part 1!)
   ├─ HyperLogLog: 2% error, 1000x less memory
   ├─ Count-Min: Over-estimate, fixed memory
   └─ Sliding Window: Bucketing reduces precision

3. HIDDEN COSTS OF REAL-TIME
   ├─ Always-on infrastructure ($$)
   ├─ State management complexity
   ├─ Fault tolerance overhead
   ├─ Late data handling
   └─ Coordination across systems

4. EVOLUTION: BATCH → STREAM → EDGE
   1960s: Store all, query later
   2026: Process at edge, aggregate globally
   └─ Each step: traded storage for speed

5. 2026 SYSTEMS ARE HYBRID
   ├─ Streaming for real-time (approximate)
   ├─ Batch for accuracy (exact)
   ├─ Lambda architecture (both!)
   └─ Edge for latency (distributed)

---

🗺️ Your Streaming Journey

Where you are now:
✓ Time/space trade-offs (Part 1)
✓ Algorithm design (Part 2)
✓ Graphs (Part 3)
✓ Production systems (Part 4)
✓ Database internals (Part 5)
✓ Caching layers (Part 6)
✓ Real-time streaming (Part 7) ← YOU ARE HERE

Your growing insight:
├─ See hidden costs everywhere
├─ Understand approximation trade-offs
├─ Can design for infinite data
├─ Know when real-time is worth the cost
└─ Ready for AI/ML algorithms!

Next steps:
□ Part 8: AI/ML algorithms (recommendations, LLMs)
□ Part 9: Security & cryptography
□ Part 10: Autonomous systems

---

💬 Your Turn

Build these yourself:

1. Implement HyperLogLog and test accuracy vs set size
2. Build sliding window counter with different bucket sizes
3. Create Count-Min Sketch and compare to HashMap
4. Measure memory: Exact vs Approximate

What would you ask?

• "What's the hidden cost of real-time?"
• "Why is approximate good enough?"
• "When would you NOT use streaming?"

Share your findings! What's your error rate vs memory trade-off? 📊

---

Infinite data needs finite memory. Streaming algorithms are the bridge. Master approximation, and you master real-time systems. 📡✨

---

🎯 Coming Up Next: Part 8

AI & Machine Learning Algorithm Engineering

From counting data to learning from it:
├─ How recommendation algorithms work
├─ Transformer attention mechanism (ChatGPT)
├─ Vector similarity at scale
├─ Online learning & bandit algorithms

Same principles: Hidden costs, trade-offs, real-time!

Stay tuned! 🤖