Partitioning¶

14:20. Design review. Someone proposes partitioning the 10 TB SaaS events table by region, because "that's how the business thinks about it." You have 20 workers, each holds about 500 GB, and product wants p95 latency per customer for yesterday.

Before you weigh in: does partitioning by region help the BI scan that filters WHERE region = 'eu-west-1', the GROUP BY customer_id rollup, both, or neither? And what happens to the one worker holding eu-west-1 if that region is 60% of traffic?

That's the whole question this page answers: how you assign work so twenty machines don't all read the same 10 TB, and so GROUP BY customer_id is even possible. Kafka topics, Spark shuffles, Iceberg files, ClickHouse parts, Cassandra vnodes: different files, same idea. If you only remember one idea from Phase 0, remember this one.

Start with the situation¶

Events:

{timestamp, customer_id, user_id, service, endpoint, region, latency_ms, status_code, bytes}

Two jobs share the lake:

1. BI scan: WHERE date = yesterday AND region = 'eu-west-1' — wants to skip data.
2. Per-customer rollup: GROUP BY customer_id — wants all rows for one customer together.

Those jobs want different keys. Partitioning is the compromise you write down, not a property of the data.

The same tension appears elsewhere:

• Observability: partition by time for “last hour,” but a noisy cluster_id is a hot shard.
• E-commerce CDC: partition Kafka by order_id so per-order history is ordered; a flash-sale SKU still hotspots inventory topics.
• IoT: partition by device_id for per-device state; a warehouse full of devices in one site_id wrecks site-level partitions.
• Fraud: partition by user_id is useless for “accounts sharing this device.”

Why the obvious approach breaks at scale¶

On one machine, “split the file into 8 chunks” is enough. On a cluster:

• Every record must belong to exactly one partition of a given scheme (or you double-count).
• Independent partitions can run in parallel; dependent ones (all rows for cust_0042) must meet — that meeting is a shuffle.
• The slowest partition is the job. See Distributed Execution.
• Changing the scheme later rewrites the world. Iceberg/Hudi rewrites files; Kafka needs a new topic; Spark pays a full shuffle.

Build the mental picture¶

A partition is a slice that a worker can own.

graph LR
    subgraph "10 TB dataset"
        P0["P0 500 GB"]
        P1["P1 500 GB"]
        P2["P2 500 GB"]
        P3["P3 500 GB"]
        Px["… 20 slices"]
    end
    W1[Worker 1] --> P0
    W2[Worker 2] --> P1
    W3[Worker 3] --> P2
    W4[Worker 4] --> P3

Two different “partitions” get confused in reviews. Keep them separate:

A well-laid-out lake still repartitions in memory when the compute key disagrees with the folder key. That is normal. Paying that shuffle and scanning 10 TB because you forgot date= is not.

"Partition" is one of the most overloaded words in this academy — the same term names at least seven different mechanisms across the systems you will touch:

Two of these are worth calling out because they are routinely confused inside the same system: ClickHouse's PARTITION BY decides which coarse-grained parts exist (and can be dropped/TTL'd together); ORDER BY decides the physical row order and sparse index inside each part. A query can prune partitions and still scan every granule in the surviving ones if ORDER BY doesn't match the filter — the two mechanisms answer different questions and neither substitutes for the other.

Mental test: If I delete one partition, did I delete a coherent business slice (a day, a tenant, a Kafka shard) or a random 128 MB? Coherent slices are how you retry and how you expire data.

Under the hood¶

Hash partitioning¶

partition = hash(key) % num_partitions

All events for customer-001 land together. Per-customer aggregations and keyed Flink state become local. Hashing does not preserve range locality: “customers A–C” is not a folder you can skip.

Spark’s default shuffle is hash (or hash + sort in the sort-shuffle writer). Kafka’s default partitioner hashes the message key.

Range partitioning¶

• P0: timestamp 2024-01-01 → 2024-03-31
• P1: 2024-04-01 → 2024-06-30

Natural for time series and for Iceberg hidden partitioning on days. Queries with BETWEEN skip files. Writes concentrate on “now” — every producer hits today’s range.

Time / identity partitioning (Hive-style)¶

/events/date=2024-01-15/hour=10/
/events/date=2024-01-15/hour=11/

This is range partitioning with a path convention. Combined with Parquet, it is the default lake layout for the SaaS table.

Composite keys¶

(date, customer_id) in storage: prune days, then hash customers inside the day. (customer_id, hour) as a compute key: splits a whale tenant across hours.

Partition pruning¶

-- Without date layout: 10 TB scan
SELECT count(*) FROM events WHERE date = '2024-01-15';

-- With date identity partitions: ~27 GB if volume is flat over a year
SELECT count(*) FROM events WHERE date = '2024-01-15';

The engine reads metadata (Hive metastore, Iceberg manifests, Spark InMemoryFileIndex), not the data, to drop files. Filters that do not match the partition column (WHERE endpoint = '/v2/events') do not prune.

How many partitions?¶

Rules of thumb (compute): 2–4× cores at the shuffle, 100–300 MB per Spark partition after compression. Storage: one day (or hour) per partition, files 128–512 MB inside it after compaction.

Kafka: throughput ≈ min(partitions, consumer threads) for a group; ordering is per partition. See Kafka partitions.

Put it to work¶

Storage layout for the SaaS lake¶

from pyspark.sql import functions as F

events = spark.read.json("s3://raw/events/")  # landing zone, not the lake

curated = (
    events
    .withColumn("date", F.to_date("timestamp"))
    .withColumn("hour", F.hour("timestamp"))
)

# Identity partitions for prune-friendly BI. Compact files, don't explode.
(
    curated
    .repartition(200, "date")   # control file count; don't use 200 × 365
    .write
    .mode("append")
    .partitionBy("date")
    .parquet("s3://analytics/events/")
)

Compute: size the shuffle¶

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", str(128 * 1024 * 1024))
spark.conf.set("spark.sql.shuffle.partitions", "400")  # starting point, AQE coalesces

p95 = events.groupBy("customer_id").agg(F.expr("percentile_approx(latency_ms, 0.95)"))

Detect a hot key before it pages you¶

from pyspark.sql import functions as F

# Cheap sketch: top tenants by rows (and bytes if you have them)
events.groupBy("customer_id").count().orderBy(F.desc("count")).show(20)

# Distribution of partition sizes after a hash split
hashed = events.withColumn("p", F.pmod(F.hash("customer_id"), 200))
hashed.groupBy("p").count().agg(
    F.min("count"), F.expr("percentile_approx(count, 0.5)"), F.max("count")
).show()

If max / median is > 5, you have a reducer problem, not a “need more executors” problem.

Salting a whale tenant¶

from pyspark.sql.functions import col, concat, lit, floor, rand, split, sum as Fsum, count

SALTS = 16
salted = events.withColumn(
    "k",
    concat(col("customer_id"), lit("#"), floor(rand() * SALTS).cast("int").cast("string")),
)
partial = salted.groupBy("k").agg(count("*").alias("n"), Fsum("bytes").alias("bytes"))
final = (
    partial
    .withColumn("customer_id", split("k", "#")[0])
    .groupBy("customer_id")
    .agg(Fsum("n").alias("n"), Fsum("bytes").alias("bytes"))
)

Two-phase aggregation: first shuffle is balanced; second shuffle is tiny. Percentiles need a different trick (sample, t-digest merge, or isolate the whale into its own job).

Production gotchas¶

The key choice table¶

The right key is a checklist¶

1. Write distribution — will ingest be uniform across values this hour?
2. Query filters — do they match the storage key so pruning works?
3. Cardinality — enough distinct values for the parallelism you need, not 10× more files than cores.
4. Hotspot risk — can one value dominate? If yes, plan salt / isolated pipeline now.
5. Ordering — Kafka/Flink: order is per partition. If you need per-order FIFO, the key is order_id, not country.

How it fails¶

graph LR
    subgraph "Hash by customer_id"
        P0["P0 BigCorp 4 TB"]
        P1["P1 SmallCo 50 GB"]
        P2["P2 MidCorp 200 GB"]
    end
    subgraph "Wall time"
        W1["Worker 1: 4 hours"]
        W2["Worker 2: 3 min"]
        W3["Worker 3: 12 min"]
    end
    P0 --> W1
    P1 --> W2
    P2 --> W3

Fixes, in order: measure the histogram → change storage key if pruning is wrong → salt / isolate whales if compute is skewed → compact files if you have a million 2 MB objects.

How to investigate¶

# Spark: confirm prune happened
events.filter("date = '2024-01-15'").explain("formatted")
# FileScan should list only that path / partition filters

Simulation: Kafka partitions if you want to feel consumer parallelism before you pay for a cluster.

Scale¶

Partition count should grow with bytes, not with “number of microservices.” Doubling workers without doubling useful partitions just means each worker is idle waiting on the same fat tasks.

Trade-offs¶

Repartitioning cost: Spark shuffle; Kafka new topic + dual-write or messy partition increase; Iceberg rewrite. Choose the storage key as if you will live with it for two years.

Alternatives¶

• Clustering / Z-order / sort_order inside a coarse partition (Iceberg, Delta, ClickHouse ORDER BY) when you need extra skip without extra directories.
• Separate tables for whales (events_whale, events_longtail) instead of heroic salting forever.
• Serving stores (KV, search, graph) when the access pattern is point-get or multi-hop — partitioning a lake will never make “friends of friends” cheap. See Fraud architecture.
• Single-node with table partitions (PARTITION BY RANGE (day) in Postgres) until scale says otherwise.

How to apply at work¶

In every design review, force these answers onto the wiki:

1. What is the storage partition key? The compute key?
2. Histogram of that key (p50 / p95 / max bytes) — from production, not intuition.
3. Cardinality, and file/partition count after a day of writes.
4. Can one value dominate? Which tenant/device/SKU?
5. Do query WHERE clauses match the storage key?
6. What happens when workers double — do partitions double, or do fat tasks just move?
7. What is the rewrite plan if we change the key in 18 months?

If nobody can paste a histogram, you do not have a partitioning strategy. You have a hope.

Check your understanding¶

You are designing a Kafka topic for e-commerce order events. Orders come from 50 countries. Volume is roughly proportional to GDP (US + EU + India dominate). Primary consumer today is “orders per country per hour.” A second consumer in Q3 wants per-customer order history in event order.

1. Trade-offs of partitioning by country?
2. By order_id?
3. By customer_id?
4. What key would you choose for consumer 1 only?
5. What do you choose if both consumers must be first-class, and what extra piece (topic, table, or job) do you add?
6. How do you stop a Black Friday SKU from hot-spotting a related inventory_updates topic?