Skip to main content

​
Data Partitioning & Sharding

When data outgrows a single machine, you need to partition (shard) it across multiple nodes. This module covers the strategies, trade-offs, and production patterns for effective data distribution. Think of partitioning like organizing a library: you could put all books on one giant shelf (single node), but eventually you run out of space and it takes too long to find anything. Instead, you split books across rooms — fiction in Room A, non-fiction in Room B, reference in Room C. The challenge is choosing the split so that most visitors only need to visit one room, and no single room is overwhelmed while others sit empty.
Module Duration: 10-14 hours
Key Topics: Hash Partitioning, Range Partitioning, Consistent Hashing, Rebalancing, Hot Spots
Interview Focus: Partition key selection, secondary indexes, cross-partition queries

​
Why Partition?

​
Partitioning Strategies

​
Hash Partitioning

​
Range Partitioning

​
Compound Key Partitioning

​
Consistent Hashing

Must-Know Algorithm: Consistent hashing is foundational for distributed caching, databases, and load balancing. Expect to explain and implement it in interviews.

​
Basic Consistent Hashing

​
Virtual Nodes

​
Implementation

​
Secondary Indexes

​
Local Secondary Indexes

​
Global Secondary Indexes

​
Rebalancing Strategies

​
Fixed Number of Partitions

​
Dynamic Partitioning

​
Advanced: Geo-Partitioning & Data Sovereignty

At Staff/Principal level, partitioning isn’t just about performance—it’s about compliance and locality. As laws like GDPR (Europe) and CCPA (California) become stricter, where data lives physically becomes a legal requirement.

​
1. Locality-Aware Partitioning

In a global database (e.g., CockroachDB, Spanner, or YugabyteDB), you can pin specific partitions to specific geographic regions.
  • The Goal: Keep data close to the user to minimize speed-of-light latency and comply with data residency laws.
  • The Mechanism: Using “Placement Policies” or “Partitioning Tags.”

​
2. Follow-the-Workload (Dynamic Migration)

Sophisticated systems can dynamically move partitions based on the time of day or access patterns.
  • Daylight Shifting: Move active user data to the “waking” hemisphere to ensure local low-latency access.
  • Access Locality: If a user from Germany suddenly starts accessing a “US East” partition frequently, the system can migrate that partition (or a replica) to “EU West” automatically.

​
3. Data Sovereignty Challenges

When data is pinned to a region, Cross-Region Operations become extremely complex:
  • Global Joins: Joining a “pinned” German user with a “pinned” Japanese order requires cross-continental network hops (200ms+).
  • Global Secondary Indexes: If the index is global, writing to the German partition might require updating an index node in the US, breaking sovereignty if the index contains PII (Personally Identifiable Information).
Staff Tip: When designing for Global Scale, “Data Residency” is often a harder constraint than “Queries Per Second.” Always ask: “Does this record have a legal right to leave this continent?”

​
Hot Spots and Skew

​
Identifying Hot Spots

​
Hot Spot Mitigation

​
Cross-Partition Operations

​
Scatter-Gather Pattern

​
Cross-Partition Transactions

​
Interview Practice

Question: How would you partition Twitter’s tweet storage?Requirements Analysis:
  • Queries: Get user’s tweets, home timeline, search
  • Scale: 500M tweets/day, 200M users
Approach:
Question: A celebrity with 100M followers posts. How do you handle the hot spot?Solution:
Question: How do you rebalance partitions without downtime?Process:

​
Advanced Design Scenarios

​
Scenario 1: Designing Partitioning for an E‑Commerce Platform

You are designing the data model and partitioning for a large e-commerce platform (catalog, carts, orders, inventory, users, payments). Key workloads:
  • Product browsing and search
  • User-specific operations (cart, wishlist, recommendations)
  • Order placement and history
  • Inventory management per warehouse/region
Partitioning strategy:
  • Users, carts, wishlists: Partition by user_id (hash or consistent hash)
    • “Get cart” and “get wishlist” → single partition
    • Cross-user queries (analytics) use separate OLAP or stream to a warehouse
  • Orders: Partition by user_id or order_id (with correlation to user)
    • Materialize read models for “orders by user” and “orders by status” using CQRS/Event Sourcing
  • Inventory: Partition by (warehouse_id, product_id)
    • Allows local, warehouse-level operations and avoids cross-partition inventory transactions
  • Search: Separate inverted index service (Elasticsearch / OpenSearch) with its own sharding
Trade-offs:
  • Many OLTP queries become single-partition and fast
  • Cross-entity queries (“top selling products globally”) handled via denormalized tables or separate systems
  • Consistency for cross-partition operations handled via Sagas/2PC as needed

​
Scenario 2: Multi-Tenant SaaS Partitioning

You are building a multi-tenant SaaS where each tenant has its own users, projects, and data. Goals:
  • Strong data isolation between tenants
  • Ability to move tenants between clusters or regions
  • Ability to put large tenants on dedicated resources
Partitioning options:
  • Database-per-tenant:
    • Pros: Strong isolation, easy per-tenant backup/restore
    • Cons: Operational overhead when tenants = 10k+
  • Shared database, tenant_id in every row:
    • Pros: Easier to operate at scale
    • Cons: Harder to move tenants, noisy neighbor effects
  • Hybrid:
    • Small tenants: shared database with partitioning by tenant_id
    • Large tenants: own database or shard, with routing table
Implementation pattern:
  • Maintain a Tenant Catalog mapping tenant_id → {cluster, database, partition_key}
  • Route each request through a Tenant Router that picks the correct connection before query execution
  • For large tenants, use more partitions or dedicated clusters and update catalog accordingly

​
Scenario 3: Time-Series Platform (Metrics/Logs)

You are designing a time-series platform that ingests telemetry, metrics, or logs from millions of devices. Characteristics:
  • Very high write throughput (append-only)
  • Queries are often time-bounded and per device / per stream
  • Data retention and tiering (hot vs cold storage)
Partitioning strategy:
  • Primary key: (tenant_id, stream_id, time_bucket)
  • Partition by hash(tenant_id, stream_id); time is part of clustering key
  • Create time buckets (e.g., daily or hourly) to allow range scans and efficient TTL/compaction
Hot partitions problem:
  • If you partition by raw timestamp, “now” is a hot partition
  • Instead, spread by (tenant_id, stream_id) and keep time in the clustering key
Storage tiering:
  • Recent buckets stored on fast nodes (SSD, high IOPS)
  • Older buckets moved to cheaper storage (object store) with precomputed aggregates

​
Key Takeaways

Partition Key is Critical

Choose partition key carefully—it determines data distribution, query efficiency, and hot spot risk.

Consistent Hashing Enables Scale

Virtual nodes provide even distribution and minimal data movement when nodes change.

Secondary Indexes Have Trade-offs

Local = fast writes, slow reads. Global = fast reads, eventually consistent writes.

Avoid Cross-Partition Operations

Design data model to keep related data together. Cross-partition = expensive.

​
Next Steps

Data Systems

Deep dive into Spanner, Cassandra, DynamoDB internals

Transactions

Handle cross-partition transactions with 2PC and Saga

​
Interview Deep-Dive

Strong Answer:
  • This is a fundamental tension in partitioning design. Hash partitioning gives you excellent point lookups (hash the key, go directly to one partition) and even data distribution, but destroys range scan ability because adjacent keys are scattered across partitions. Range partitioning preserves key order (enabling efficient range scans) but creates hot spots when access patterns cluster around certain key ranges.
  • The solution is usually a compound partition key. For example, in a time-series system, partition by hash(device_id) and use timestamp as the clustering key within each partition. Point lookups by device_id go to one partition, and range scans for “device X, last 24 hours” are also single-partition because the timestamps are ordered within the partition.
  • DynamoDB models this explicitly with its partition key (hash distributed) and sort key (range ordered within the partition). Cassandra uses a similar compound primary key model.
  • The design rule: pick the partition key to match your most frequent query pattern, then use secondary indexes or materialized views for alternative access patterns.
Follow-up: What if your most frequent query requires a range scan across a dimension that cannot be the sort key?Then you need a global secondary index (GSI) or a denormalized table. With a GSI (as in DynamoDB), the index is itself partitioned by the indexed attribute, so a range scan on that attribute becomes a single-partition read on the index. The cost: writes are slower because every data write triggers an asynchronous index update, and the index is eventually consistent with the base table. Alternatively, you can create a separate denormalized table optimized for the range scan query, maintained via change data capture (CDC) or an event stream. This is the CQRS pattern — separate read models for separate query patterns. The trade-off is storage cost and the complexity of keeping multiple representations in sync.
Strong Answer:
  • This is the “celebrity problem” or “thundering herd on a single key.” The partition holding the celebrity’s data receives 1000x normal traffic, exceeding its capacity while other partitions sit idle.
  • Immediate mitigation: add a random suffix to the celebrity’s partition key, splitting their data across N sub-partitions (e.g., “celebrity123:shard_0” through “celebrity123:shard_99”). Writes are distributed randomly across sub-shards. Reads require scatter-gather across all sub-shards, but this is parallelizable and much better than overloading a single partition.
  • Detection must be automated. Monitor per-partition QPS and latency in real time. When a partition’s QPS exceeds a threshold (e.g., 5x the average), automatically add it to a “hot key” list and enable sub-sharding for that key. This is a dynamic, not static, process — celebrities come and go.
  • Caching is the complementary strategy. For read-heavy hot spots, place the celebrity’s data in a distributed cache (Redis Cluster, Memcached) with a short TTL. This absorbs the read storm without hitting the database. For write-heavy hot spots (likes, view counts), buffer writes in memory and flush to the database in batches (write coalescing).
Follow-up: How do you handle the transition when a hot key is detected and you need to add sub-sharding without downtime?The transition must be atomic from the client’s perspective. I would use a routing layer that maintains a mapping of hot keys to their sub-shard configuration. When a key is detected as hot: (1) The routing layer starts writing to sub-shards in addition to the original key (dual-write phase). (2) After a convergence period, all recent data is in sub-shards. (3) Switch reads to scatter-gather across sub-shards. (4) Stop writing to the original key. The critical detail is that during the transition, reads must check both the original key and the sub-shards to avoid missing data. A simpler approach, used by some systems: always use sub-sharding for all keys (with a low shard count like 4), so no dynamic transition is needed. The overhead of reading 4 sub-shards per key is minimal, and you are always prepared for hot keys.
Strong Answer:
  • A local secondary index is co-located with the data partition. Each partition maintains its own index over its local data only. Writes are fast because the index update is a local operation within the same partition. But reads by the indexed attribute require scatter-gather across all partitions because any partition might have matching data. Read latency equals the slowest partition.
  • A global secondary index is separately partitioned by the indexed attribute. A query like “find all users in NYC” goes to a single index partition, making reads fast. But writes are slow because inserting a record might require updating an index partition on a different node, which is either a distributed transaction (expensive) or eventually consistent (the index lags behind the data).
  • Cassandra uses local indexes (materialized views with scatter-gather). DynamoDB uses global indexes (GSIs that are eventually consistent). Elasticsearch uses global indexing with near-real-time consistency.
  • The decision depends on read/write ratio. If the workload is write-heavy and reads by secondary key are rare, local indexes are better. If the workload is read-heavy with frequent secondary key lookups, global indexes are better despite the write overhead.
Follow-up: In DynamoDB, what consistency issues can arise from the eventually consistent nature of GSIs?A write to the base table is immediately visible but the GSI update is asynchronous, typically within milliseconds but potentially delayed during high write throughput. This means a query on the GSI might not return a record that was just written to the base table. Worse, it can return stale attribute values if the record was recently updated. In practice, this means you cannot use a GSI query result as the source of truth for an immediate business decision. For example, if you use a GSI to check “is this username taken?” and the GSI lags, two users might simultaneously register the same username — the GSI returns “not taken” for both. The mitigation: use the base table (with a strongly consistent read) for invariant checks, and use GSIs only for queries where eventual consistency is acceptable (search, analytics, listing pages).