Unit II: Mining Data Streams — The Fun Way!

"A stream of data is like a river — you can't step in the same river twice, and you definitely can't store the whole river in a bucket."

What You'll Learn (Table of Contents)

1. Introduction to Stream Concepts
2. Stream Data Model and Architecture
3. Stream Computing
4. Sampling Data in a Stream
5. Filtering Streams
6. Counting Distinct Elements in a Stream
7. Estimating Moments
8. Counting Oneness in a Window
9. Decaying Window
10. Real Time Analytics Platform (RTAP) Applications
11. Case Studies

• Real Time Sentiment Analysis
• Stock Market Predictions

1. Introduction to Stream Concepts

Data Streams: The River That Never Stops

Simple Definition: A data stream is a continuous, never-ending flow of data items arriving in real-time — from things like sensors, user clicks, financial trades, or social media posts.

4 Key Characteristics of Streams

1. Unboundedness — "Are We There Yet?" (No, Never)

Analogy: It's like trying to take a shower but the water never turns off. You can enjoy it, but you have to keep choosing what to do with the water now.

2. High Velocity — It Arrives FAST

• Thousands to millions of events per second
• Twitter: ~6,000 tweets/second on an average day, spikes to 150,000/second during major events
• NASDAQ stock exchange: millions of price ticks per second during trading hours
• A modern car's ECU: 25,000+ sensor readings per second

Formula 1 Analogy: A Formula 1 car has ~300 sensors sending 1,500 data points per second. The team's engineers are doing real-time stream analytics on live race data to advise the driver. If they used batch processing, the race would be over before they finished analyzing Lap 1!

3. One-Pass Constraints — You Get ONE Look

Live TV Analogy: Watching a live cricket match vs. watching a recording. During the live match, you can't rewind. You process what you see, when you see it!

4. Approximate Answers — "Close Enough" is a Feature, Not a Bug

• Exact calculations on massive streams require too much memory
• If exact answer takes 10 seconds and approximate answer takes 10ms, the approximate answer is 1,000x more useful in real-time systems
• For most business decisions, "~50,000 unique visitors" is as actionable as "50,247 unique visitors"

Weather Analogy: When you check a weather app and it says "29°C", it's actually 28.7°C. The rounding is an approximation. Do you need more precision to decide whether to wear a jacket? No! The approximate answer is perfectly useful.

Real-World Stream Examples

Illustrative Example: A network monitoring system receives a stream of packet headers. It:

• Filters suspicious IPs (filtering)
• Counts distinct source addresses (distinct counting)
• Alerts when an unusual spike occurs (anomaly detection)

ALL of this happens in milliseconds, on data it can never fully store!

2. Stream Data Model and Architecture

The "Conveyor Belt Factory" Model

 Raw Ingredients → Loading Dock → Processing → Packaging → Shelves
 (Data Sources) (Kafka/Kinesis) (Flink/Storm) (Storage) (Dashboard)

The Full Architecture Blueprint

Each Layer Explained

Sources / Producers — The Origin of Everything

• IoT devices: Temperature sensors, smart meters, GPS trackers
• Applications: Your app sending click events, page views, errors
• External feeds: Stock price APIs, weather feeds, social media APIs

The producer's job: Keep generating data as fast as possible and hand it off.

Ingestion Layer — The Post Office

• Accepts data from producers
• Stores messages durably for a configurable time (hours, days, weeks)
• Allows multiple consumers to read independently at their own pace
• Guarantees messages aren't lost even if downstream systems are slow

Analogy: Kafka is like a post office with a filing cabinet. The postman (producer) drops letters. Different departments (consumers) collect their own copies at their own pace. The filing cabinet holds letters for 7 days before discarding them.

Processing Engines — The Brains

Flink vs Storm: Storm is faster but Flink is smarter. Storm processes one event at a time (true streaming). Flink has richer windowing, exactly-once semantics, and better state management. For most new projects, choose Flink!

Storage / Sinks — Where Results Land

• Redis → For fast, real-time lookups (e.g., "Is this user online?", fraud scores)
• Cassandra → For time-series data requiring high write throughput
• Elasticsearch → For searchable analytics and log analysis
• HDFS/S3 → For archival and batch processing later

The Stream as a Math Sequence

(timestamp: 2024-01-01T10:00:00, user_id: u001, event: "click", page: "/home")

STREAM → [FILTER operator] → [AGGREGATE operator] → [ALERT operator] → OUTPUT

3. Stream Computing

Two Types of Stream Operators

Stateless Operators — "No Memory, No Problem"

Analogy: A stateless operator is like a rubber stamp — every letter gets the same treatment, regardless of what came before.

Stateful Operators — "I Remember Everything"

Memory Analogy: A stateful operator is like a bank teller who remembers your account balance from previous transactions, not just your current request.

Windowing — The Key Concept in Stream Computing

3 Types of Windows:

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ TIME →

 TUMBLING WINDOW (fixed, non-overlapping):
 [──────10min──────][──────10min──────][──────10min──────]
 Window 1 Window 2 Window 3
 Each event in EXACTLY one window.

 SLIDING WINDOW (overlapping):
 [────10min────]
 [────10min────]
 [────10min────]
 Slide every 5 min. Events appear in MULTIPLE windows.

 SESSION WINDOW (activity-based):
 [──activity──] gap [─────activity─────] gap [─act─]
 Window closes after X minutes of inactivity.

Fault Tolerance — "What if a Server Dies Mid-Stream?"

1. Periodically save the current state to durable storage (HDFS, S3)
2. If a node crashes, restore from last checkpoint
3. Replay events that arrived after the checkpoint (Kafka holds them!)

Analogy: Checkpointing is like the "Save Game" feature. If your game crashes, you restart from the last save point, not from the very beginning!

• At-least-once: Every event is processed, but some may be processed twice (duplicates possible)
• Exactly-once: Every event is processed exactly once, even after failures (Flink's superpower! )

4. Sampling Data in a Stream

The "Why Not Process Everything?" Problem

Analogy: A chef doesn't eat the entire pot of soup to check if it's tasty — they taste one spoonful. If the sample is representative, it tells you about the whole pot!

Reservoir Sampling — The Classic Algorithm

Step 1: Fill the reservoir with the first k items.
Step 2: For item i (where i > k):
 - Generate random number j between 1 and i
 - If j ≤ k: replace reservoir[j] with item i
 - Else: discard item i
Step 3: Repeat until stream ends.

Casino Analogy: Imagine a raffle where each new ticket buyer has a chance to replace a previously selected winner. As more people enter, the replacement probability drops — but everyone has a FAIR chance overall!

Poisson Sampling — The Simple Alternative

for each element in stream:
 if random() < p:
 include in sample
 else:
 skip

• Super simple to implement
• Parallelizable (each worker samples independently)
• Sample size is random (expected $p \cdot n$, but varies)
• Good for distributed systems where exact sample size isn't critical

Stratified Sampling — Fair Representation

Poll Analogy: If you're surveying 1000 people about politics, random sampling might give you 950 from one city and 50 from everywhere else. Stratified sampling ensures all regions are proportionally represented.

Real Example: E-Commerce Transaction Sampling

An e-commerce site processes 2 million orders per hour. The data science team needs a sample to train a fraud detection model quickly.

1. Use reservoir sampling with $k = 50,000$
2. Result: Statistically valid sample from all 2 million orders
3. Train the fraud model on the sample (faster!)
4. Deploy the model to score ALL 2 million orders in real-time

5. Filtering Streams

The Bouncer at the Club

Core Idea: Apply a predicate (condition) to each element. If TRUE → keep it. If FALSE → drop it.

Why Filtering Matters — The Numbers

• Without filtering early: Your expensive ML model processes 1,000,000 events/sec
• With filtering: Your ML model processes 10,000 events/sec (100x less work!)

Factory Analogy: A quality control inspector at the START of the assembly line catches and discards defective parts early. This prevents expensive downstream processing on bad parts. Same idea!

Types of Filters

Simple (Stateless) Filters:

-- SQL-style filter (Flink SQL)
WHERE temperature > 50.0 -- Value threshold
WHERE country = 'IN' -- Specific key
WHERE message LIKE '%error%' -- Pattern match
WHERE transaction_amount > 1000 -- Business rule

Stateful Filters (More Powerful!):

Only pass events WHERE the condition has been TRUE for 3 consecutive readings

Example: Alert only if temperature > 50°C for 3 consecutive readings (not just a single spike). This prevents false alarms from noisy sensors!

Where to Filter — Push It as Early as Possible!

Real Example: Stock Ticker Filtering

The NYSE emits every trade (millions per day). An analyst only cares about large trades (block trades > 10,000 shares) for market impact analysis.

Filter rule: trade.volume > 10,000 AND trade.exchange = 'NYSE'

DataStream<Trade> largeTrades = allTrades
 .filter(trade -> trade.getVolume() > 10_000 
 && trade.getExchange().equals("NYSE"));

6. Counting Distinct Elements in a Stream

The Impossible-But-Essential Problem

• 10M unique users × 8 bytes per ID = 80 MB of memory
• With 1 billion users: 8 GB just for one counter!

Flajolet-Martin (FM) Sketch — The Magic of Bit Tricks

The Trick: Hash each element and look at how many trailing zeros appear in the binary representation of the hash. The more zeros, the rarer it is — and the rarity tells us the total count!

• About 50% of hashes end with at least 1 trailing zero
• About 25% end with at least 2 trailing zeros
• About 12.5% end with at least 3 trailing zeros
• If the MAXIMUM trailing zeros you've ever seen is $r$, then roughly $2^r$ unique elements exist!

Element "Alice" → hash = 10110100 → trailing zeros = 2 → 2^2 = 4
Element "Bob" → hash = 11001000 → trailing zeros = 3 → 2^3 = 8
Element "Carol" → hash = 10100110 → trailing zeros = 1 → 2^1 = 2
Element "Dave" → hash = 01110000 → trailing zeros = 4 → 2^4 = 16

Max trailing zeros seen = 4 → Estimate = 2^4 = 16 unique elements
(Actual = 4, but we'd average multiple hash functions for better accuracy)

HyperLogLog — The Industry Standard

• Redis (built-in PFADD, PFCOUNT commands)
• Google BigQuery (APPROX_COUNT_DISTINCT)
• Amazon Redshift, Snowflake (approximate count functions)

1. Hash each element
2. Divide into $m$ registers (buckets) based on first few bits
3. Track the maximum trailing zeros seen in each register
4. Use the harmonic mean of $2^{\text{max_zeros}}$ across all registers as your estimate

Mind-blowing fact: HyperLogLog can estimate cardinality of 1 BILLION unique elements with just 1.5 KB of memory and ±2% error. The naïve approach needs 8 GB!

Mergeability — The Superpower of Sketches

Server 1: Processed events 1-100,000 → HLL sketch A
Server 2: Processed events 100,001-200,000 → HLL sketch B

Merged sketch = MERGE(A, B) → Estimates unique elements across ALL 200,000 events!

This allows distributed counting without a central coordinator!

7. Estimating Moments

What's a "Moment" and Why Do We Care?

0th Moment — Count of Distinct Elements

1st Moment — Easy Running Sum

total = 0
for element in stream:
 total += element.value # or total += 1 for count

2nd Moment — The AMS Sketch

• $F_2$ is low when all items appear roughly equally often (uniform)
• $F_2$ is high when a few items dominate (skewed)

Real application: Detecting "heavy hitters" in network traffic. If a few IP addresses are generating disproportionate traffic, $F_2$ will be large → potential DDoS attack!

Analogy: Imagine randomly assigning +1 or -1 to each person's "score" and summing. Most scores cancel out, but the dominant contributors (people with very high scores) survive the averaging — and that tells you about the distribution.

Visual: How Moments Reveal Distribution

Low F₂ (Uniform — everyone contributes equally):
████████████████████████████████████
Each item: ████ ████ ████ ████ ████

High F₂ (Skewed — a few items dominate):
████████████████████████████████████████████████
One item: ████████████████████████████████████████████████
Others: ██ █ ██ █ ██

8. Counting Oneness in a Window

"How Many 1s in the Last 60 Seconds?"

Why binary streams? Many real signals are binary:

• Did a user click? (1) or not? (0)
• Did a transaction get flagged? (1) or not? (0)
• Did the sensor trigger an alert? (1) or not? (0)

Exact Method — The Simple Queue

window = deque() # holds timestamps of 1s
for item, timestamp in stream:
 if item == 1:
 window.append(timestamp)
 # Remove expired entries
 while window and timestamp - window[0] > window_size:
 window.popleft()
 count_of_ones = len(window)

The DGIM Algorithm — Approximate with Logarithmic Memory

• Divide the stream into "buckets"
• Each bucket has a size that's a power of 2: 1, 2, 4, 8, 16, ...
• Keep at most 2 buckets of each size
• When you get 3 buckets of the same size, merge them into one bucket of double size

Stream: 1 0 1 1 0 1 0 0 1 1 0 1 ...
 ↑ current time

Buckets (size, timestamp_of_most_recent_1):
 [1, t=12] [1, t=11] [2, t=9] [4, t=6] [8, t=2]

Analogy: Instead of remembering every single 1 you've seen, you remember "groups" of 1s using less and less precise timestamps for older groups — like how you remember "I had coffee sometime last week" but precisely remember "I had coffee at 9:15 AM today."

Window Types Revisited — Now With More Depth

9. Decaying Window

"Recent News Matters More Than Old News"

News Analogy: A tweet about a product crash posted 5 minutes ago should influence sentiment scores WAY more than a tweet from 6 months ago. A decaying window naturally handles this!

The Math Behind Decay

• $\lambda$ is the decay rate (larger $\lambda$ = faster decay = shorter "memory")
• $e^{-\lambda t}$ approaches 0 as $t$ grows

λ = 0.1 (slow decay, long memory):
 10 seconds ago: e^(-0.1×10) = 0.37 (still 37% weight)
 60 seconds ago: e^(-0.1×60) = 0.002 (almost gone)

λ = 1.0 (fast decay, short memory):
 5 seconds ago: e^(-1.0×5) = 0.007 (almost gone!)
 10 seconds ago: e^(-1.0×10) = 0.00005 (essentially 0)

Implementation — The Elegant Trick

aggregate = 0
last_time = 0
λ = 0.1

for value, current_time in stream:
 # Decay the existing aggregate
 time_delta = current_time - last_time
 aggregate = aggregate * math.exp(-λ * time_delta)

 # Add new value
 aggregate += value
 last_time = current_time

 print(f"Decaying aggregate: {aggregate:.3f}")

Choosing the Right Decay Rate

Analogy: Think of $\lambda$ as the "memory slider" on a mixing board. All the way up = only care about RIGHT NOW. All the way down = treat all history equally (basically no decay).

Real Application: Social Media Sentiment Tracker

Negative tweet at t=0: contribution = -1.0 × e^(-0.01×0) = -1.0
Positive tweet at t=30s: contribution = +1.0 × e^(-0.01×30) = +0.74
Negative tweet at t=31s: contribution = -1.0 × e^(-0.01×31) = -0.73
Positive tweet at t=120s: contribution = +1.0 × e^(-0.01×120) = +0.30

Current decaying sentiment = -1.0 + 0.74 - 0.73 + 0.30 = -0.69 → Mostly negative!

10. Real Time Analytics Platform (RTAP) Applications

When "Later is Too Late"

Emergency Room Analogy: In an ER, triage must happen within seconds. A doctor who reviews patient symptoms "sometime this afternoon" is useless if the patient arrived 3 hours earlier. RTAPs are the ER triage of data systems!

The RTAP Stack

Application 1: Fraud Detection

Your card is swiped in Mumbai at 2 PM
 ↓
Kafka event: {card: "4532...", amount: ₹5000, merchant: "Electronics Store", lat: 19.07, lon: 72.87, time: "14:00:00"}
 ↓
Flink stream processor receives event in <1ms
 ↓
Feature extraction:
 - Last 7 days spending average: ₹1200
 - This transaction is 4.2x above average 
 - Location matches home city 
 - Device fingerprint matches 
 - Merchant category: NORMAL 
 ↓
ML model score: 0.23 (threshold 0.80 for block)
 ↓
Decision: APPROVE (but log as slightly suspicious)

Application 2: IoT Anomaly Detection

• Collect data → Wait for daily batch job → Report broken machines the next morning
• Problem: Machine already failed at 3 AM, caused 8 hours of downtime!

Sensor stream: temperature, vibration, pressure readings every 100ms
 ↓
Sliding window (5 minutes): compute mean and standard deviation
 ↓
Z-score = (current_value - mean) / std_dev
If Z-score > 3σ → ANOMALY DETECTED 
 ↓
Alert sent to maintenance team in < 2 seconds!

Result: Instead of 8 hours of downtime, maintenance crew arrives within 10 minutes of anomaly detection. Savings: potentially hundreds of thousands of rupees per incident!

Application 3: Social Media Monitoring

Incoming tweet stream filtered for brand mentions
 ↓
NLP model assigns sentiment: positive / neutral / negative
(using spaCy or BERT-based model inside Flink)
 ↓
Sliding 5-minute window: count negative mentions
 ↓
If negative_count > 500 in 5 minutes → SPIKE ALERT! 
 ↓
Marketing team notified via Slack + PagerDuty

Without RTAP: Brand discovers the PR disaster 24 hours later in a morning report. Already viral, stock price dropped 2%!

With RTAP: Marketing team responds within 10 minutes. Crisis contained!

Application 4: IoT Smart City Dashboard

• Traffic sensors at 10,000 intersections
• Air quality monitors at 500 stations
• Smart meters for electricity usage
• GPS on 2,000 public transport vehicles

Result: City officials can reroute traffic, adjust traffic light timing, and send pollution alerts in real-time — not based on yesterday's batch report!

11. Case Studies

Case Study 1: Real-Time Sentiment Analysis

The Problem

The Architecture

Step-by-Step Pipeline

• Twitter Firehose: up to 500M tweets/day
• Filtered to brand mentions: ~50,000/day (99% dropped at source!)
• Kafka buffers these with guaranteed delivery

Rules:
- Language = English 
- Contains brand keyword 
- NOT a retweet 
- Account age > 30 days (spam filter) 
- Not duplicate tweet ID 

• Pre-trained BERT model fine-tuned on product reviews
• Runs inside Flink as a UDF (user-defined function)
• Output: {tweet_id, sentiment: "positive/neutral/negative", score: 0.92, topic: "product_quality"}
• Latency per tweet: ~15ms

Sliding window: 5 minutes, slide: 1 minute
Compute:
 - Average sentiment score
 - Count by sentiment category
 - Count by topic
 - Count by region

IF negative_count_5min > 200 AND avg_sentiment < -0.6:
 TRIGGER: "Sentiment Spike Alert"
 → Notify: PR team (Slack)
 → Notify: Customer service (PagerDuty)
 → Create: Incident ticket (Jira)

Why This Matters

Real example: When United Airlines faced a viral PR crisis (passenger removed forcibly, 2017), the story spread throughout the night before their PR team responded. Estimated stock market damage: $1 billion in market cap the next day. RTAP could have enabled faster crisis management.

Case Study 2: Stock Market Predictions

The Environment

• NYSE processes ~1 billion transactions/day
• Each tick: milliseconds apart
• Competition uses algorithms, not humans
• 1 millisecond advantage = millions in profits (or losses prevented)

"Latency arms race": Some firms spend millions co-locating their servers in the same building as exchange servers — just to save 1 millisecond of network latency!

The Pipeline

Key Technical Features

Simple Moving Average (SMA) — Tumbling window:
 SMA_20 = average price over last 20 candles (20 min window)

Exponential Moving Average (EMA) — Decaying window:
 EMA gives more weight to recent prices (λ parameter tuned per asset)

Relative Strength Index (RSI) — Sliding window:
 RSI = 100 - (100 / (1 + avg_gain_14 / avg_loss_14))
 RSI > 70 = overbought, RSI < 30 = oversold

Bollinger Bands:
 Upper band = SMA + 2 × standard_deviation
 Lower band = SMA - 2 × standard_deviation

RULES:
- Position size ≤ 2% of total portfolio
- Stop-loss at 0.5% loss per trade
- Max drawdown today ≤ 5%
- Halt all trading if confidence < 60%

The Latency Challenge

Target: Decision in < 10 milliseconds end-to-end!

Timeline breakdown:
 Market feed to Kafka: ~0.5ms (network)
 Kafka to Flink: ~0.5ms (disk + network)
 Feature computation: ~2ms (CPU)
 Model inference: ~3ms (CPU/GPU)
 Risk check: ~1ms (Redis lookup)
 Order submission: ~2ms (network to exchange)
 ─────────────────────────
 Total: ~9ms 

For context: Human reaction time is ~200ms. Algorithmic trading at 9ms is 22x faster than a human blink!

Appendix: Advanced Stream Topics

Backpressure and Flow Control

Kafka: ████████████████████ producing at 1M events/sec
Flink: ████████ processing at 500K events/sec
Gap: ████████ 500K events/sec backing up!

1. Kafka consumer pause: Flink tells Kafka "slow down!" when buffers fill up
2. Auto-scaling: Add more Flink workers (Kubernetes horizontal pod autoscaling)
3. Load shedding: Drop low-priority events during overload (last resort!)

Plumbing Analogy: Backpressure is like the pressure relief valve on a water heater. When pressure gets too high (buffer full), it releases some (pauses consumption) to prevent an explosion (data loss).

Security & Compliance in Streams

• PII (names, emails, phone numbers)
• Financial data (card numbers, transaction amounts)
• Health data (vitals, diagnoses)

Example: A fintech company streams transaction data through Kafka. Before hitting analytics, a Flink operator tokenizes card numbers: 4532xxxxxxx → TKN-a7f3b2c1. Analytics sees tokens, never real card numbers. Even if someone intercepts the stream, they get useless tokens!

State Management in Distributed Streams

Every 10 seconds:
 1. Pause processing momentarily (milliseconds!)
 2. Write current state to HDFS/S3 as "checkpoint"
 3. Resume processing

If Machine #3 crashes:
 1. Restore its state from last checkpoint
 2. Replay Kafka messages from the checkpoint offset
 3. Back online! (Exactly-once semantics preserved)

Unit Summary

Test Your Knowledge — Quick Quiz!

• A small business with 10,000 users — 80 KB is fine
• When you need to know if a specific user appeared, not just the count
• When 2% error is unacceptable (billing systems, legal compliance)

• 3:01 PM (60 sec later): The 3 PM spike still contributes $e^{-0.01 \times 60} \approx 0.55$ of its original weight. Still quite visible.
• 4:00 PM (3600 sec later): The 3 PM spike contributes $e^{-0.01 \times 3600} \approx 0.000000015$ — essentially zero. The tracker has completely "forgotten" the 3 PM spike.

1. The last checkpoint saved the state at time $T$
2. Kafka retains messages from before $T$ (replay capability)
3. Flink restores state from the checkpoint
4. Flink replays all Kafka messages from offset at time $T$ onwards
5. With exactly-once semantics, idempotent writes ensure no duplicates reach the sink

Mnemonics to Remember

Further Reading & Resources