Partitioning

​

Introduction

​

What is partitioning?

​

Why partition data?

1. Scalability - Handle more data
   • Single machine has limited disk capacity (maybe 1-10TB)
   • With partitioning, dataset can grow to petabytes by adding more machines
   • Example: Facebook has petabytes of user data, impossible to store on one machine
2. Performance - Handle more requests
   • Single machine has limited CPU and memory
   • Query throughput limited by single machine (e.g., 10,000 queries/second)
   • With partitioning, queries distributed across many machines
   • If you have 10 partitions, theoretically 10x throughput
   • Example: Twitter handles millions of tweets/second by partitioning across thousands of machines
3. Parallel query processing
   • Large queries can be parallelized across multiple partitions
   • Each partition processes its subset of data independently
   • Results combined at the end
   • Example: Analytics query “count users by country” - each partition counts its users, results aggregated

​

Partitioning vs. replication

• Partitioning: Divide data into subsets
• Replication: Keep multiple copies of each partition for fault tolerance

​

Partitioning of key-value data

​

Partitioning by key range

class RangePartitioner:
    def __init__(self, boundaries):
        # boundaries = ['E', 'M', 'S']  means:
        # Partition 1: A-E, Partition 2: F-M, Partition 3: N-S, Partition 4: T-Z
        self.boundaries = sorted(boundaries)

    def get_partition(self, key):
        # Determine which partition a key belongs to
        for i, boundary in enumerate(self.boundaries):
            if key <= boundary:
                return i
        return len(self.boundaries)  # Last partition

# Example usage
partitioner = RangePartitioner(['E', 'M', 'S'])
print(partitioner.get_partition('Alice'))    # 0 (Partition 1: A-E)
print(partitioner.get_partition('Mary'))     # 1 (Partition 2: F-M)
print(partitioner.get_partition('Tom'))      # 3 (Partition 4: T-Z)

1. Range queries are efficient
   • If you want all users from Alice to Charlie, you know they’re all in Partition 1
   • No need to query all partitions
   • Example: “Get all temperature readings between timestamp 2024-01-01 and 2024-01-31”
2. Keys are stored in sorted order within partition
   • Easier to scan through records in order
   • Good for applications that need sorted data

# Instead of just timestamp as key
key = timestamp

# Use composite key
key = f"{sensor_id}:{timestamp}"
# Now data distributed across sensors, not just time

• BigTable/HBase: Keys stored in sorted order within each partition
• MongoDB (before 2.4): Could partition by range (now recommends hash-based)
• RethinkDB: Supports range-based sharding

​

Partitioning by hash of key

class HashPartitioner:
    def __init__(self, num_partitions):
        self.num_partitions = num_partitions

    def get_partition(self, key):
        # Hash the key and mod by number of partitions
        # hash() returns an integer
        return hash(key) % self.num_partitions

# Example usage
partitioner = HashPartitioner(4)
print(partitioner.get_partition('Alice'))    # e.g., 2
print(partitioner.get_partition('Bob'))      # e.g., 0
print(partitioner.get_partition('Charlie'))  # e.g., 3
print(partitioner.get_partition('David'))    # e.g., 1

• A good hash function takes skewed data and makes it uniformly distributed
• Same input always produces same output (deterministic)
• No need for cryptographically strong hash (MD5, SHA-256 overkill)
• Common choices: MurmurHash, CityHash, FNV

# Python's built-in hash() is NOT suitable for partitioning!
# It may give different results in different processes

# BAD - Don't use for distributed systems
partition = hash(key) % num_partitions  # ❌

# GOOD - Use consistent hash function
import hashlib
partition = int(hashlib.md5(key.encode()).hexdigest(), 16) % num_partitions  # ✓

1. Even distribution of data
   • Hash function uniformly distributes keys across partitions
   • Reduces risk of hot spots
   • Example: User IDs hashed - users evenly distributed regardless of naming patterns
2. Automatic load balancing
   • No manual adjustment of partition boundaries needed
   • Works well even if data access patterns change

1. Range queries are inefficient
   • Adjacent keys in key space end up in different partitions
   • Example: “Get all users from Alice to Charlie” requires querying ALL partitions
2. Loss of ordering
   • Hash destroys the natural ordering of keys
   • Can’t efficiently iterate through keys in sorted order

• Cassandra: Uses consistent hashing (hash of key determines partition)
• MongoDB: Default sharding strategy uses hash of shard key
• Riak: Uses consistent hashing
• DynamoDB: Hash partitioning

​

Consistent hashing

# With 4 partitions
partition = hash(key) % 4  # Key "Alice" → partition 2

# Add 5th partition
partition = hash(key) % 5  # Key "Alice" → partition 4 (moved!)

# Almost all keys move to different partitions!

1. Hash output space forms a ring (e.g., 0 to 2^32-1, wraps around)
2. Each partition assigned a position on the ring
3. Each key hashed to a position on the ring
4. Key belongs to the next partition clockwise on the ring

• Only keys between new partition and previous partition need to move
• Other keys stay in same partition

​

Skewed workloads and hot spots

# Social media platform partitioned by user_id
# User ID: celebrity_123 (millions of followers)

# All writes to celebrity's posts go to one partition!
celebrity_partition = hash('celebrity_123') % num_partitions

# Millions of users reading celebrity's posts
# All reads go to same partition → HOT SPOT! 🔥

1. Application-level sharding:
   # Add random suffix to celebrity's posts
   celebrity_id = 'celebrity_123'
   random_suffix = random.randint(0, 99)
   key = f"{celebrity_id}_{random_suffix}"
   
   # Now celebrity's posts spread across multiple partitions
   # When reading, query all suffixes and merge
   
2. Caching layer:
   • Put cache (Redis, Memcached) in front of hot partitions
   • Cache absorbs read load
   • Database partition sees less traffic
3. Read replicas for hot partition:
   • Create more replicas of the hot partition
   • Distribute reads across replicas

​

Partitioning and secondary indexes

-- Primary key: car_id
CREATE TABLE cars (
    car_id INT PRIMARY KEY,
    make VARCHAR(50),
    model VARCHAR(50),
    color VARCHAR(50),
    price DECIMAL
);

-- Easy query (by primary key)
SELECT * FROM cars WHERE car_id = 12345;

-- Hard query (by secondary attribute)
SELECT * FROM cars WHERE color = 'red' AND make = 'Tesla';

​

Document-based partitioning (local indexes)

class DocumentPartitionedDB:
    def __init__(self, num_partitions):
        self.partitions = [Partition(i) for i in range(num_partitions)]

    def insert(self, car_id, make, color, price):
        # Partition by primary key
        partition = self.get_partition(car_id)

        # Insert document
        partition.insert(car_id, {'make': make, 'color': color, 'price': price})

        # Update local indexes
        partition.index_color[color].append(car_id)
        partition.index_make[make].append(car_id)

    def query_by_color(self, color):
        # Must query ALL partitions (scatter/gather)
        results = []
        for partition in self.partitions:
            # Each partition searches its local index
            car_ids = partition.index_color.get(color, [])
            results.extend([partition.get(id) for id in car_ids])
        return results

• Fast writes: Only need to update one partition
• Simple: Each partition is independent

• Slow reads: Need to query all partitions (scatter/gather)
• Tail latency problem: Read as slow as the slowest partition
• If one partition is slow or unavailable, entire query affected

• MongoDB: Local secondary indexes
• Elasticsearch: Each shard has its own indexes
• Cassandra: Local secondary indexes (added in version 2.1)
• Riak: Search (based on Solr) uses local indexes

​

Term-based partitioning (global indexes)

• Global index partitioned by the indexed field (e.g., color)
• color: a-m → Index Partition 0
• color: n-z → Index Partition 1
• All red cars (regardless of car_id) → same index partition

• Fast reads: Only need to query one index partition
• More efficient for read-heavy workloads
• Better performance for queries on secondary attributes

• Slow writes: Must update multiple partitions (data + indexes)
• Complex distributed transactions needed
• Often updated asynchronously (eventual consistency)

• DynamoDB: Global Secondary Indexes (GSIs)
• Riak Search: Can use global indexes
• Oracle: Global indexes in partitioned tables

• Writes faster (don’t wait for index updates)
• But: Reads may not immediately see new data
• Eventually consistent (index catches up)

​

Comparison: local vs. global indexes

​

Rebalancing partitions

• Data size increases: More data doesn’t fit in existing partitions
• Query load increases: Need more machines to handle traffic
• Machines fail: Need to redistribute load to surviving machines
• New machines added: Want to take advantage of additional resources

​

Requirements for rebalancing

1. After rebalancing, load should be shared fairly between nodes
   • Data and query load distributed evenly
   • No hot spots created
2. Database should continue accepting reads and writes during rebalancing
   • No downtime
   • Minimal performance impact
3. No more data than necessary should be moved between nodes
   • Moving data is expensive (network bandwidth, disk I/O)
   • Minimize disruption

​

Strategies for rebalancing

​

Don’t hash mod N (bad approach)

# Initial: 3 nodes
partition = hash(key) % 3

# Add 4th node
partition = hash(key) % 4

# Almost EVERY key moves to different partition!

​

Fixed number of partitions

• Create fixed number of partitions (e.g., 1000 partitions)
• Each node owns several partitions
• When new node added: Steal a few partitions from existing nodes
• Partitions themselves don’t change, just reassigned to different nodes

• Only entire partitions moved (clear boundaries)
• Can move partition while continuing to serve reads/writes (replica takes over)
• Number of partitions doesn’t change

• Riak: 64 partitions per node (if cluster has 10 nodes → 640 partitions)
• Elasticsearch: Shards created upfront, can’t change without reindexing
• Couchbase: 1024 vBuckets per bucket

​

Dynamic partitioning

• Number of partitions adapts to data volume
• Works well with both key-range and hash partitioning
• Empty database starts with small number of partitions, grows organically

• Empty database starts with single partition → all writes to one node (hot spot)
• Pre-splitting: Some databases allow configuring initial set of partitions

• HBase: Automatic partition splitting
• MongoDB: Automatic chunk splitting (64MB chunks)
• RethinkDB: Dynamic sharding

​

Partitioning proportionally to nodes

# Each node gets fixed number of partitions (e.g., 128)
num_partitions = num_nodes * 128

# Add node: Number of partitions increases
# Remove node: Number of partitions decreases

• Automatically adapts to cluster size
• Each new node takes fair share of load

• Cassandra: Uses vnodes (virtual nodes), default 256 per physical node
• Riak: Also uses vnodes

​

Comparison of rebalancing strategies

​

Automatic vs. manual rebalancing

• System automatically decides when and how to move partitions
• Convenient, less operational burden
• Risk: Can go wrong (move too much data, overwhelm network, cause cascading failures)

• Human administrator decides partition assignment
• System executes the move
• More control, prevents surprises
• Requires more operational effort

​

Request routing

​

Approaches to request routing

​

Approach 1: Client contacts any node

• Client doesn’t need to know cluster topology
• Simple client logic
• Any node can handle any request (after forwarding)

• Extra network hop if first node doesn’t have data
• All nodes need routing table

• Cassandra: Uses this approach (gossip protocol shares routing info)
• Riak: Similar approach

​

Approach 2: Routing tier

• Client completely unaware of partitioning
• Routing logic centralized (easier to update)
• Nodes don’t need to forward requests

• Routing tier is additional component (single point of failure unless replicated)
• Extra network hop

• MongoDB: Uses mongos routers (routing tier)

​

Approach 3: Client-side routing

• Most efficient (no extra hops)
• No additional routing infrastructure needed

• Client needs to be partition-aware
• Client needs to track cluster changes
• More complex client logic

​

How does routing learn about partition changes?

​

Approach: Coordination service (e.g., ZooKeeper)

1. Each node registers itself in ZooKeeper with partition assignments
2. Routing tier (or client) subscribes to ZooKeeper for updates
3. When partitions reassigned, ZooKeeper notifies subscribers
4. Routing tier updates its routing table

• HBase: Uses ZooKeeper for metadata
• Kafka: Uses ZooKeeper (moving away from it in newer versions)
• SolrCloud: Uses ZooKeeper

​

Approach: Gossip protocol

1. Every node periodically picks random node to share state with
2. Information spreads through cluster exponentially fast
3. Eventually all nodes have consistent view of cluster

• Cassandra: Uses gossip protocol
• Riak: Also uses gossip

​

Summary

• Key range: Efficient range queries, but risk of hot spots
• Hash: Even distribution, but inefficient range queries
• Consistent hashing: Minimal rebalancing when nodes added/removed

• Local (document-based): Fast writes, slow reads (scatter/gather)
• Global (term-based): Fast reads, slow writes (distributed updates)

• Fixed partitions: Simple, but need to choose number upfront
• Dynamic partitions: Adapts to data size, but complex
• Proportional to nodes: Adapts to cluster size

• Client to any node (with forwarding)
• Routing tier (partition-aware load balancer)
• Client-side routing (partition-aware client)
• Use ZooKeeper or gossip for coordination