Streaming & Big Data

Why "big" data, and when it isn't

Data is "big" when a single machine can't store it, can't process it fast enough, or can't keep up with its arrival rate. Then you scale out: split the data into partitions and process them on many machines in parallel. Distributed systems pay real costs for that: network transfer, serialization, coordination, partial failures, and much harder debugging.

Meanwhile, single machines have grown enormous. A laptop has 8–16 cores, 16–64 GB of RAM, and an NVMe SSD reading several GB/s; a cloud VM can have hundreds of cores and terabytes of RAM. Columnar, vectorized, multi-threaded engines (DuckDB, Polars) stream data from disk and spill to disk when memory runs out, so they comfortably process hundreds of gigabytes on one node.

A year of a busy clickstream, compressed in Parquet, is around 800 GB, and a typical query touches a month of it, a few columns at a time. That is single-node territory. Reach for a cluster when the data is in the tens of terabytes and queries touch most of it, when many teams share one platform, or when the arrival rate exceeds what one machine can ingest.

Remember: estimate compressed size and the fraction a query touches before choosing a cluster.

Streaming concepts and Kafka

An event is an immutable fact about something that happened, with a timestamp: "user 42 viewed page /pricing at 10:00:03". A stream is an unbounded, ordered sequence of events. Producers append events; consumers read them. A broker such as Apache Kafka (or Redpanda, Amazon Kinesis, Google Pub/Sub, Apache Pulsar) sits between them, durably storing events so producers and consumers are decoupled in time and speed.

Topics, partitions, and offsets

Kafka organizes events into topics (for example clicks). Each topic is split into partitions, and each partition is an append-only log: events get consecutive offsets 0, 1, 2… Ordering is guaranteed within a partition, not across partitions. A producer chooses the partition by hashing the event's key, so all events with the same key (the same user) land in the same partition, in order. Events are retained for a configured time (say seven days) whether or not anyone has read them, so consumers can replay.

Consumer groups and committed offsets

A consumer group is a set of consumers sharing the work of reading a topic. Kafka assigns each partition to exactly one consumer in the group, so the number of partitions caps the group's parallelism (four partitions, six consumers: two sit idle). Each group tracks its own committed offset per partition, the position it has processed up to, so different groups (a dashboard, a fraud detector, an archiver) read the same topic independently. When a consumer joins or dies, the group rebalances and partitions move.

Watch records move through a streaming pipeline: each stage handles a record as soon as the previous stage passes it on, so all stages work at once.

Remember: topics split into partitions; partitions are ordered logs with offsets; keys pick partitions; a consumer group shares partitions and commits offsets.

Windowing with generators

You can't "sum all events" in an unbounded stream: it never ends. Instead, aggregate over windows, finite slices of time:

Generators are a natural fit: a stream is an iterator that may never end, and a windowing operator is a generator that consumes events and yields each window result as soon as the window closes, holding only the current window's state in memory.

Remember: tumbling = fixed and disjoint; sliding = fixed and overlapping; session = per key, closed by inactivity. Emit a window when it closes, and keep only open windows in memory.

Event time, late data, and watermarks

Every event has two times: event time (when it happened, stamped by the device) and processing time (when your system sees it). They differ, sometimes by hours: a phone in airplane mode uploads its clicks on landing; a network partition delays a whole data center. Windowing by processing time is easy but wrong for most analytics ("sales between 10:00 and 10:01" means event time).

Windowing by event time raises a question: when can you close the 10:00 window, if a 10:00 event might still arrive? A watermark is the system's estimate that "all events with event time ≤ W have probably arrived". A common heuristic is max event time seen minus an allowed lateness. When the watermark passes a window's end, the window is emitted. Events that arrive for an already-closed window are late: drop them, send them to a side output for correction, or update the emitted result (which downstream must handle).

Remember: window by event time; the watermark decides when a window closes; late events need a policy.

Backpressure and exactly-once

Backpressure is how a slow consumer tells a fast producer to slow down. Without it, the gap between them accumulates in memory until the process crashes. The simplest mechanism is a bounded buffer: when it's full, the producer blocks (or drops, or samples). Kafka consumers get backpressure for free because they pull: a slow consumer simply falls behind, and its consumer lag (latest offset minus committed offset) grows, which is the key metric to alert on.

Exactly-once, honestly

"Exactly-once" in streaming means the effects of each event appear once, despite retries and crashes. It is achieved with one of:

• Idempotent sinks: at-least-once delivery plus upserts keyed by event id (the M62 approach; simplest and most common).
• Transactions: Kafka's idempotent producer plus transactions let a consume-transform-produce loop commit its output records and input offsets atomically (isolation.level=read_committed on readers).
• Checkpointed state: Flink and Spark snapshot operator state and source offsets together, and on failure rewind both to the last checkpoint.

Each guarantee covers only the systems inside the transaction or checkpoint. Sending an email or calling an external API from a stream processor is at-least-once unless that API is idempotent (an idempotency key).

Remember: bounded buffers and pull-based consumers give backpressure; watch consumer lag; exactly-once is end-to-end engineering, usually via idempotent writes.

Distributed processing: MapReduce

MapReduce (Google, 2004; Hadoop) is the model under every distributed engine. A job has three phases:

1. Map: each worker reads one input split and emits (key, value) pairs, independently and in parallel.
2. Shuffle: pairs are routed by key (hash of the key mod the number of reducers) so all values for a key meet at the same reducer. This is the expensive phase: data crosses the network and is sorted.
3. Reduce: each reducer combines the values for each of its keys.

flowchart LR
  S1[split 1] --> M1[map]
  S2[split 2] --> M2[map]
  S3[split 3] --> M3[map]
  M1 --> SH{{shuffle by hash of key}}
  M2 --> SH
  M3 --> SH
  SH --> R1[reduce: keys a-m]
  SH --> R2[reduce: keys n-z]
  R1 --> O[(output)]
  R2 --> O
    

Remember: map is embarrassingly parallel; the shuffle moves data by key and dominates cost; combiners shrink it.

Dask and Spark

Modern engines generalize MapReduce into a DAG of operations and add lazy evaluation: transformations only build a plan; nothing runs until an action (compute(), collect(), count(), a write) asks for a result. Laziness lets the optimizer fuse steps, push filters down to the file scan, and prune unneeded columns, just like DuckDB and Polars lazy mode.

Dask

Dask scales the Python APIs you know. A Dask DataFrame is many pandas DataFrames (partitions); Dask Array is many NumPy chunks; dask.delayed turns arbitrary functions into a lazy task graph. It runs on a laptop's cores or a cluster with the same code.

Apache Spark and PySpark

Spark is the most widely used distributed engine. A driver program builds a plan; a cluster manager starts executors on many machines; the plan is split into stages at every shuffle boundary, and each stage into one task per partition.

• RDD (resilient distributed dataset): the original low-level API, a distributed collection of Python objects transformed with functions (map, reduceByKey). Flexible but opaque to the optimizer, and Python functions force serialization between the JVM and Python.
• DataFrame: a distributed table with a schema. Operations are declarative, so the Catalyst optimizer rewrites the plan and the Tungsten engine executes it in the JVM, much faster. Use DataFrames (or Spark SQL) by default.
• Narrow vs wide transformations: filter, select, withColumn are narrow (each output partition depends on one input partition, no data movement). groupBy, join, distinct, orderBy are wide: they need a shuffle and start a new stage.

Remember: transformations are lazy, actions run; shuffles (wide transformations) define stages and cost the most; prefer DataFrames to RDDs.

Partitioning and skew

Parallelism only helps if work is spread evenly. Engines assign rows to partitions by hash partitioning (hash(key) % n, used by shuffles), range partitioning (sorted key ranges, used for sorting), or the input file layout. Skew happens when a few keys hold most of the data (one celebrity account, one giant customer, a NULL key): one task gets far more rows than the others, and the whole stage waits for that straggler.

Other remedies: broadcast joins (send a small table to every worker instead of shuffling the big one; Spark's F.broadcast(df)), Spark 3's adaptive query execution (AQE), which splits skewed partitions automatically, and filtering or handling NULL keys separately. On disk, aim for partitions (files) of roughly 100 MB–1 GB; thousands of tiny files are a form of skew in the other direction ("small files problem"), fixed by compaction.

Remember: look at the task-duration distribution; one straggler means skew; salt, broadcast, or let AQE split it.

Cloud object storage

Big-data systems store their files in object storage: Amazon S3, Google Cloud Storage, Azure Blob Storage, or S3-compatible systems (MinIO, Cloudflare R2). It is cheap, effectively unlimited, durable (S3 is designed for eleven nines), and readable by every engine. But it is not a filesystem, and its differences shape data layouts:

• Buckets and keys: a flat namespace of objects addressed by key, such as s3://spokes-lake/trips/year=2026/month=03/part-0.parquet. "Folders" are just key prefixes; listing a prefix is an API call that pages through keys.
• Immutable objects: you write a whole object or replace it; there's no append and no cheap rename (a "rename" is copy plus delete). That is why table formats (Iceberg, Delta) commit by writing new files plus a new metadata pointer instead of renaming.
• Request costs and latency: each GET/PUT/LIST costs money and tens of milliseconds, so many small files are slow and expensive. Large files plus range reads (Parquet footers let readers fetch only needed column chunks) work best.
• Consistency: S3 has offered strong read-after-write consistency since December 2020, so a successful PUT is immediately visible to GET and LIST.
• Multipart uploads for large objects, storage classes (Standard, Infrequent Access, Glacier) with lifecycle rules, and IAM policies for access control.

Remember: object storage is cheap, durable, and flat; write large immutable files under partition-style prefixes; let a table format handle commits.

Practice

def tumbling(events, size):
    current, count, total = None, 0, 0
    for ts, amount in events:
        start = int(ts // size) * size
        if current is not None and start != current:
            yield current, count, total
            count, total = 0, 0
        current = start
        count += 1
        total += amount
    if current is not None:
        yield current, count, total

from collections import defaultdict

def sessionize(events, gap):
    by_user = defaultdict(list)
    for ts, user in events:
        by_user[user].append(ts)
    out = []
    for user, times in by_user.items():
        times.sort()
        start = prev = times[0]
        n = 1
        for t in times[1:]:
            if t - prev > gap:
                out.append((user, start, prev, n))
                start, n = t, 0
            prev = t
            n += 1
        out.append((user, start, prev, n))
    return sorted(out)

import re
import zlib
from collections import defaultdict

def map_phase(text):
    return [(w, 1) for w in re.findall(r"[a-z]+", text.lower())]

def shuffle(pairs, n):
    buckets = [defaultdict(list) for _ in range(n)]
    for word, value in pairs:
        buckets[zlib.crc32(word.encode()) % n][word].append(value)
    return [dict(b) for b in buckets]

def reduce_phase(bucket):
    return {word: sum(values) for word, values in bucket.items()}

def word_count(texts, n=3):
    pairs = [p for t in texts for p in map_phase(t)]
    result = {}
    for bucket in shuffle(pairs, n):
        result.update(reduce_phase(bucket))
    return result

def window_with_watermark(events, size, lateness):
    open_, closed, late = {}, [], []
    watermark = float("-inf")
    for t, value in events:
        start = t // size * size
        if start + size <= watermark:
            late.append((t, value))
            continue
        open_[start] = open_.get(start, 0) + value
        watermark = max(watermark, t - lateness)
        for s in sorted(w for w in open_ if w + size <= watermark):
            closed.append((s, open_.pop(s)))
    for s in sorted(open_):
        closed.append((s, open_[s]))
    return closed, late

import zlib
from collections import Counter

def partition_sizes(keys, n):
    sizes = [0] * n
    for k in keys:
        sizes[zlib.crc32(k.encode()) % n] += 1
    return sizes

def skew(sizes):
    mean = sum(sizes) / len(sizes)
    return 0.0 if mean == 0 else max(sizes) / mean

def salt(keys, hot, k):
    seen = Counter()
    out = []
    for key in keys:
        if key in hot:
            out.append(f"{key}#{seen[key] % k}")
            seen[key] += 1
        else:
            out.append(key)
    return out

Mini-project: a streaming clickstream aggregator

pixi init clickstream && cd clickstream
pixi add python structlog pytest
pixi run python -m clickstream --minutes 10 --late-rate 0.05 --seed 1
pixi run pytest
# stretch: a real broker
pixi add python-confluent-kafka

Quiz

Recap

• Estimate compressed size and the fraction queries touch; single-node DuckDB and Polars handle hundreds of gigabytes.
• Kafka topics split into partitions, ordered logs with offsets; keys choose partitions; consumer groups share partitions and commit offsets.
• Tumbling, sliding, and session windows turn unbounded streams into finite results; generators implement them with bounded memory.
• Window by event time; watermarks decide when windows close; late data needs an explicit policy.
• Bounded buffers give backpressure; exactly-once means idempotent sinks, transactions, or checkpointed state.
• MapReduce = map, shuffle, reduce; Dask and Spark add lazy DAGs; shuffles are the expensive wide operations.
• Skew creates stragglers: salt hot keys, broadcast small tables, or use AQE. Object storage favours large, immutable, partitioned files.

Going further

• asyncio queues and concurrent.futures in the Python docs
• Apache Kafka documentation and confluent-kafka for Python
• Dask and PySpark documentation
• Tyler Akidau, "Streaming 101" and "Streaming 102", and the book Streaming Systems
• Martin Kleppmann, Designing Data-Intensive Applications, chapters 10–11
• Dean and Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters" (2004)
• Next: put it all together in Project 9 · ETL Pipeline & Warehouse