IH

What is database sharding and how do you implement it?

Answer

Database sharding is a horizontal partitioning technique that splits a large database into smaller, more manageable pieces called shards. Each shard is a separate database that contains a subset of the total data, distributed across multiple servers to improve performance, scalability, and availability.

Sharding Strategies

1. Range-Based Sharding

-- Shard by date ranges
-- Shard 1: Orders from 2023
CREATE TABLE orders_2023 (
    order_id BIGINT PRIMARY KEY,
    customer_id INT,
    order_date DATE,
    total DECIMAL(10,2)
) -- Stored on Server 1

-- Shard 2: Orders from 2024
CREATE TABLE orders_2024 (
    order_id BIGINT PRIMARY KEY,
    customer_id INT,
    order_date DATE,
    total DECIMAL(10,2)
) -- Stored on Server 2

-- Application logic for routing
function getShardForDate(orderDate) {
    if (orderDate >= '2024-01-01') return 'shard_2024';
    if (orderDate >= '2023-01-01') return 'shard_2023';
    return 'shard_archive';
}

2. Hash-Based Sharding

-- Shard by customer ID hash
-- Shard determination: customer_id % 4

-- Shard 0: customer_id % 4 = 0
CREATE TABLE customers_shard_0 (
    customer_id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    email VARCHAR(100)
) -- Server 1

-- Shard 1: customer_id % 4 = 1
CREATE TABLE customers_shard_1 (
    customer_id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    email VARCHAR(100)
) -- Server 2

-- Application routing logic
function getCustomerShard(customerId) {
    return `shard_${customerId % 4}`;
}

-- Consistent hashing for better distribution
function consistentHash(key, shardCount) {
    const hash = md5(key);
    return parseInt(hash.substring(0, 8), 16) % shardCount;
}

3. Directory-Based Sharding

-- Lookup service to map data to shards
CREATE TABLE shard_directory (
    entity_type VARCHAR(50),
    entity_id VARCHAR(100),
    shard_id VARCHAR(50),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (entity_type, entity_id)
);

-- Example entries
INSERT INTO shard_directory VALUES
('customer', '12345', 'shard_west', NOW()),
('customer', '67890', 'shard_east', NOW()),
('order', 'ORD001', 'shard_west', NOW());

-- Query to find shard
SELECT shard_id FROM shard_directory 
WHERE entity_type = 'customer' AND entity_id = '12345';

Implementation Approaches

1. Application-Level Sharding

class ShardManager:
    def __init__(self):
        self.shards = {
            'shard_0': 'mysql://user:pass@db1:3306/shard_0',
            'shard_1': 'mysql://user:pass@db2:3306/shard_1',
            'shard_2': 'mysql://user:pass@db3:3306/shard_2',
            'shard_3': 'mysql://user:pass@db4:3306/shard_3'
        }
    
    def get_shard(self, customer_id):
        shard_key = f"shard_{customer_id % 4}"
        return self.shards[shard_key]
    
    def execute_query(self, customer_id, query, params=None):
        connection_string = self.get_shard(customer_id)
        conn = connect(connection_string)
        return conn.execute(query, params)
    
    def execute_cross_shard_query(self, query, params=None):
        results = []
        for shard_name, connection_string in self.shards.items():
            conn = connect(connection_string)
            result = conn.execute(query, params)
            results.extend(result)
        return results

2. Middleware-Based Sharding

-- Using ProxySQL for MySQL sharding
-- Configure sharding rules
INSERT INTO mysql_query_rules (
    rule_id, active, match_pattern, destination_hostgroup, apply
) VALUES 
(1, 1, '^SELECT.*FROM customers WHERE customer_id = ([0-9]+).*', 
 0, 1),  -- Route to appropriate hostgroup based on customer_id

-- Hostgroup configuration
INSERT INTO mysql_servers (hostgroup_id, hostname, port) VALUES
(0, 'shard0.example.com', 3306),
(1, 'shard1.example.com', 3306),
(2, 'shard2.example.com', 3306),
(3, 'shard3.example.com', 3306);

3. Database-Native Sharding

-- PostgreSQL: Foreign Data Wrappers for sharding
CREATE EXTENSION postgres_fdw;

-- Create foreign servers
CREATE SERVER shard_1 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'shard1.example.com', port '5432', dbname 'shard_db');

CREATE SERVER shard_2 FOREIGN DATA WRAPPER postgres_fdw
OPTIONS (host 'shard2.example.com', port '5432', dbname 'shard_db');

-- Create user mappings
CREATE USER MAPPING FOR current_user SERVER shard_1
OPTIONS (user 'shard_user', password 'password');

-- Create foreign tables
CREATE FOREIGN TABLE customers_shard_1 (
    customer_id INT,
    customer_name VARCHAR(100),
    email VARCHAR(100)
) SERVER shard_1 OPTIONS (schema_name 'public', table_name 'customers');

-- Create partitioned table with foreign table partitions
CREATE TABLE customers (
    customer_id INT,
    customer_name VARCHAR(100),
    email VARCHAR(100)
) PARTITION BY HASH (customer_id);

-- Attach foreign tables as partitions
ALTER TABLE customers ATTACH PARTITION customers_shard_1
FOR VALUES WITH (MODULUS 4, REMAINDER 0);

Cross-Shard Operations

1. Cross-Shard Queries

-- Aggregation across shards
-- Application-level aggregation
function getTotalOrdersByRegion() {
    const results = {};
    
    for (const shard of shards) {
        const shardResults = shard.query(`
            SELECT region, COUNT(*) as order_count, SUM(total) as total_amount
            FROM orders 
            WHERE order_date >= '2024-01-01'
            GROUP BY region
        `);
        
        // Merge results
        for (const row of shardResults) {
            if (!results[row.region]) {
                results[row.region] = { count: 0, amount: 0 };
            }
            results[row.region].count += row.order_count;
            results[row.region].amount += row.total_amount;
        }
    }
    
    return results;
}

-- Using UNION ALL for cross-shard queries
SELECT region, SUM(order_count) as total_orders, SUM(total_amount) as total_revenue
FROM (
    SELECT region, COUNT(*) as order_count, SUM(total) as total_amount
    FROM shard_1.orders WHERE order_date >= '2024-01-01' GROUP BY region
    
    UNION ALL
    
    SELECT region, COUNT(*) as order_count, SUM(total) as total_amount
    FROM shard_2.orders WHERE order_date >= '2024-01-01' GROUP BY region
    
    UNION ALL
    
    SELECT region, COUNT(*) as order_count, SUM(total) as total_amount
    FROM shard_3.orders WHERE order_date >= '2024-01-01' GROUP BY region
) combined_results
GROUP BY region;

2. Distributed Transactions

# Two-Phase Commit for cross-shard transactions
class DistributedTransaction:
    def __init__(self, shard_manager):
        self.shard_manager = shard_manager
        self.participants = []
    
    def begin_transaction(self, shard_ids):
        self.participants = []
        for shard_id in shard_ids:
            conn = self.shard_manager.get_connection(shard_id)
            conn.execute("BEGIN")
            self.participants.append((shard_id, conn))
    
    def prepare_phase(self):
        prepared_shards = []
        try:
            for shard_id, conn in self.participants:
                # Phase 1: Prepare
                result = conn.execute("PREPARE TRANSACTION 'xact_001'")
                if result.success:
                    prepared_shards.append((shard_id, conn))
                else:
                    raise Exception(f"Prepare failed on shard {shard_id}")
            return prepared_shards
        except Exception as e:
            # Abort on all prepared shards
            for shard_id, conn in prepared_shards:
                conn.execute("ROLLBACK PREPARED 'xact_001'")
            raise e
    
    def commit_phase(self, prepared_shards):
        for shard_id, conn in prepared_shards:
            # Phase 2: Commit
            conn.execute("COMMIT PREPARED 'xact_001'")

Shard Management

1. Shard Splitting

-- Split a shard when it becomes too large
-- Original shard: customer_id % 2 = 0 (customers 0, 2, 4, 6, ...)
-- Split into: 
--   customer_id % 4 = 0 (customers 0, 4, 8, ...)
--   customer_id % 4 = 2 (customers 2, 6, 10, ...)

-- Step 1: Create new shard
CREATE TABLE customers_shard_new (
    customer_id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    email VARCHAR(100)
);

-- Step 2: Migrate data
INSERT INTO customers_shard_new
SELECT * FROM customers_shard_0 
WHERE customer_id % 4 = 2;

-- Step 3: Remove migrated data from original shard
DELETE FROM customers_shard_0 
WHERE customer_id % 4 = 2;

-- Step 4: Update application routing logic
function getCustomerShard(customerId) {
    return `shard_${customerId % 4}`;  // Changed from % 2 to % 4
}

2. Shard Rebalancing

class ShardRebalancer:
    def __init__(self, shard_manager):
        self.shard_manager = shard_manager
    
    def analyze_shard_distribution(self):
        shard_stats = {}
        for shard_id in self.shard_manager.get_all_shards():
            stats = self.shard_manager.execute_query(shard_id, """
                SELECT 
                    COUNT(*) as row_count,
                    pg_size_pretty(pg_total_relation_size('customers')) as size
                FROM customers
            """)
            shard_stats[shard_id] = stats
        return shard_stats
    
    def rebalance_shards(self, source_shard, target_shard, migration_criteria):
        # Step 1: Identify data to migrate
        migration_data = self.shard_manager.execute_query(source_shard, f"""
            SELECT * FROM customers WHERE {migration_criteria}
        """)
        
        # Step 2: Insert into target shard
        for row in migration_data:
            self.shard_manager.execute_query(target_shard, 
                "INSERT INTO customers VALUES (%s, %s, %s)", 
                (row.customer_id, row.customer_name, row.email))
        
        # Step 3: Remove from source shard
        self.shard_manager.execute_query(source_shard, f"""
            DELETE FROM customers WHERE {migration_criteria}
        """)

Challenges and Solutions

1. Hot Spots

-- Problem: Uneven data distribution
-- Solution: Better shard key selection

-- Bad: Sequential IDs create hot spots
-- All new customers go to the latest shard
CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,  -- Sequential, creates hot spots
    customer_name VARCHAR(100)
);

-- Good: Use compound shard key
CREATE TABLE customers (
    customer_id BIGINT PRIMARY KEY,
    region VARCHAR(10),
    customer_name VARCHAR(100)
);

-- Shard by region + hash of customer_id
function getShardKey(customerId, region) {
    const regionHash = hash(region) % 2;  // 2 regions per shard group
    const customerHash = hash(customerId) % 2;  // 2 customer groups per region
    return `shard_${regionHash}_${customerHash}`;
}

2. Cross-Shard Joins

-- Challenge: JOINs across shards are expensive
-- Solution 1: Denormalization
CREATE TABLE order_details_denormalized (
    order_id BIGINT,
    customer_id INT,
    customer_name VARCHAR(100),  -- Denormalized from customers table
    customer_email VARCHAR(100), -- Denormalized from customers table
    product_id INT,
    product_name VARCHAR(100),   -- Denormalized from products table
    quantity INT,
    unit_price DECIMAL(10,2)
);

-- Solution 2: Co-location of related data
-- Shard both customers and orders by customer_id
function getShardForCustomerData(customerId) {
    return `shard_${customerId % 4}`;
}

-- Both tables use same sharding strategy
CREATE TABLE customers_shard_0 AS SELECT * FROM customers WHERE customer_id % 4 = 0;
CREATE TABLE orders_shard_0 AS SELECT * FROM orders WHERE customer_id % 4 = 0;

-- Now JOINs work within each shard
SELECT c.customer_name, COUNT(o.order_id) as order_count
FROM customers_shard_0 c
LEFT JOIN orders_shard_0 o ON c.customer_id = o.customer_id
GROUP BY c.customer_id, c.customer_name;

3. Global Constraints

-- Challenge: Unique constraints across shards
-- Solution: Global ID generation service

-- Centralized ID generator
CREATE TABLE global_sequences (
    sequence_name VARCHAR(50) PRIMARY KEY,
    current_value BIGINT NOT NULL,
    increment_by INT DEFAULT 1
);

-- Function to get next global ID
CREATE OR REPLACE FUNCTION get_next_global_id(seq_name VARCHAR(50))
RETURNS BIGINT AS $$
DECLARE
    next_id BIGINT;
BEGIN
    UPDATE global_sequences 
    SET current_value = current_value + increment_by
    WHERE sequence_name = seq_name
    RETURNING current_value INTO next_id;
    
    RETURN next_id;
END;
$$ LANGUAGE plpgsql;

-- Alternative: UUID-based approach
-- Use UUIDs to ensure global uniqueness without coordination
CREATE TABLE customers (
    customer_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    customer_name VARCHAR(100),
    email VARCHAR(100) UNIQUE  -- Still need to check uniqueness across shards
);

Monitoring and Maintenance

1. Shard Health Monitoring

class ShardMonitor:
    def __init__(self, shard_manager):
        self.shard_manager = shard_manager
    
    def check_shard_health(self):
        health_report = {}
        
        for shard_id in self.shard_manager.get_all_shards():
            try:
                # Check connectivity
                conn = self.shard_manager.get_connection(shard_id)
                
                # Check basic metrics
                metrics = conn.execute("""
                    SELECT 
                        (SELECT COUNT(*) FROM customers) as customer_count,
                        (SELECT COUNT(*) FROM orders) as order_count,
                        pg_database_size(current_database()) as db_size
                """).fetchone()
                
                health_report[shard_id] = {
                    'status': 'healthy',
                    'customer_count': metrics.customer_count,
                    'order_count': metrics.order_count,
                    'db_size': metrics.db_size
                }
                
            except Exception as e:
                health_report[shard_id] = {
                    'status': 'unhealthy',
                    'error': str(e)
                }
        
        return health_report

2. Performance Monitoring

-- Monitor query performance across shards
CREATE TABLE shard_performance_log (
    shard_id VARCHAR(50),
    query_type VARCHAR(50),
    execution_time_ms INT,
    rows_affected INT,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Log slow queries
CREATE OR REPLACE FUNCTION log_shard_performance(
    p_shard_id VARCHAR(50),
    p_query_type VARCHAR(50),
    p_execution_time INT,
    p_rows_affected INT
) RETURNS VOID AS $$
BEGIN
    INSERT INTO shard_performance_log 
    (shard_id, query_type, execution_time_ms, rows_affected)
    VALUES (p_shard_id, p_query_type, p_execution_time, p_rows_affected);
END;
$$ LANGUAGE plpgsql;

-- Analyze performance trends
SELECT 
    shard_id,
    query_type,
    AVG(execution_time_ms) as avg_execution_time,
    MAX(execution_time_ms) as max_execution_time,
    COUNT(*) as query_count
FROM shard_performance_log
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '1 hour'
GROUP BY shard_id, query_type
ORDER BY avg_execution_time DESC;

Best Practices

1. Shard Key Selection

  • Choose keys with good distribution (avoid hot spots)
  • Consider query patterns (minimize cross-shard queries)
  • Plan for future scaling (avoid frequent resharding)
  • Use compound keys when necessary

2. Data Consistency

  • Implement eventual consistency where possible
  • Use distributed transactions sparingly
  • Design for idempotency
  • Handle partial failures gracefully

3. Operational Considerations

  • Automate shard provisioning and scaling
  • Implement comprehensive monitoring
  • Plan for disaster recovery across shards
  • Test failover scenarios regularly

Interview Tips

  • Understand when sharding is necessary vs other scaling approaches
  • Know different sharding strategies and their trade-offs
  • Be familiar with challenges like cross-shard queries and global constraints
  • Understand the complexity of distributed transactions
  • Know how to design shard keys for even distribution
  • Practice explaining shard management operations (splitting, rebalancing)
  • Be aware of alternatives like read replicas and vertical partitioning

Test Your Knowledge

Take a quick quiz to test your understanding of this topic.