Sharding — Database Scaling

Sharding means distributing data across multiple independent database instances or servers. When a single database becomes too large or receives too much traffic, we can split the data across multiple databases called shards.

Sharding means distributing data across multiple independent database instances or servers.

When a single database becomes too large or receives too much traffic, we can split the data across multiple databases called shards.

Why Do We Need Sharding?

Suppose an e-commerce application has 10 billion orders.

Instead of storing everything in one database:

flowchart LR
    A[ASP.NET Core Web API] --> B[Single Database]
    B --> C[10 Billion Orders]

we can distribute the data:

flowchart LR
    A[ASP.NET Core Web API] --> B[Shard Router]

    B --> C[Shard 1
Customers 1-1M]
    B --> D[Shard 2
Customers 1M-2M]
    B --> E[Shard 3
Customers 2M-3M]
    B --> F[Shard 4
Customers 3M-4M]

Each shard contains only a portion of the total data.

Sharding vs Partitioning

Partitioning	Sharding
Usually inside one database	Across multiple database instances
Database manages partitions	Application/router often determines shard
One database server can still be the limit	Can distribute load across servers
Simpler operationally	More complex

Example: Customer-Based Sharding

Suppose we have three SQL Server databases:

OrdersShard1
OrdersShard2
OrdersShard3

We decide that CustomerId determines the shard.

flowchart TD
    A[CustomerId] --> B{Shard Router}

    B -->|1 - 1000000| C[SQL Server Shard 1]
    B -->|1000001 - 2000000| D[SQL Server Shard 2]
    B -->|2000001 - 3000000| E[SQL Server Shard 3]

SQL Server Complete Example

Create the same Orders table in each database.

Shard 1

CREATE DATABASE OrdersShard1;
GO

USE OrdersShard1;
GO

CREATE TABLE Orders
(
    OrderId INT PRIMARY KEY,
    CustomerId INT NOT NULL,
    Amount DECIMAL(18,2) NOT NULL,
    OrderDate DATE NOT NULL,
    Status VARCHAR(30) NOT NULL
);
GO

INSERT INTO Orders
    (OrderId, CustomerId, Amount, OrderDate, Status)
VALUES
    (1, 100100, 1000.00, '2026-09-01', 'Completed'),
    (2, 500500, 2500.00, '2026-09-02', 'Pending');
GO

Shard 2

CREATE DATABASE OrdersShard2;
GO

USE OrdersShard2;
GO

CREATE TABLE Orders
(
    OrderId INT PRIMARY KEY,
    CustomerId INT NOT NULL,
    Amount DECIMAL(18,2) NOT NULL,
    OrderDate DATE NOT NULL,
    Status VARCHAR(30) NOT NULL
);
GO

INSERT INTO Orders
    (OrderId, CustomerId, Amount, OrderDate, Status)
VALUES
    (3, 1500000, 3000.00, '2026-09-03', 'Completed'),
    (4, 1800000, 4500.00, '2026-09-04', 'Pending');
GO

Shard 3

CREATE DATABASE OrdersShard3;
GO

USE OrdersShard3;
GO

CREATE TABLE Orders
(
    OrderId INT PRIMARY KEY,
    CustomerId INT NOT NULL,
    Amount DECIMAL(18,2) NOT NULL,
    OrderDate DATE NOT NULL,
    Status VARCHAR(30) NOT NULL
);
GO

INSERT INTO Orders
    (OrderId, CustomerId, Amount, OrderDate, Status)
VALUES
    (5, 2500000, 5000.00, '2026-09-05', 'Completed'),
    (6, 2800000, 6500.00, '2026-09-06', 'Pending');
GO

ASP.NET Core Web API Routing

This is an important system-design question:

How does the ASP.NET Core Web API know which database to query?

The API can use the CustomerId from the route to determine the shard.

GET /api/customers/1500000/orders

The request reaches the API:

sequenceDiagram
    participant Client
    participant API as ASP.NET Core Web API
    participant Router as Shard Router
    participant DB as SQL Server Shard 2

    Client->>API: GET /api/customers/1500000/orders
    API->>Router: GetShard(1500000)
    Router-->>API: Shard 2
    API->>DB: SELECT Orders
    DB-->>API: Orders
    API-->>Client: JSON Response

ASP.NET Core Controller

[ApiController]
[Route("api/customers")]
public class OrdersController : ControllerBase
{
    private readonly IShardRouter _shardRouter;

    public OrdersController(IShardRouter shardRouter)
    {
        _shardRouter = shardRouter;
    }

    [HttpGet("{customerId:long}/orders")]
    public async Task GetOrders(long customerId)
    {
        var connectionString =
            _shardRouter.GetConnectionString(customerId);

        await using var connection =
            new SqlConnection(connectionString);

        await connection.OpenAsync();

        const string sql = """
            SELECT OrderId, CustomerId, Amount,
                   OrderDate, Status
            FROM Orders
            WHERE CustomerId = @CustomerId
            ORDER BY OrderDate;
            """;

        await using var command = new SqlCommand(sql, connection);

        command.Parameters.AddWithValue("@CustomerId", customerId);

        var orders = new List<object>();

        await using var reader =
            await command.ExecuteReaderAsync();

        while (await reader.ReadAsync())
        {
            orders.Add(new
            {
                OrderId = reader.GetInt32(0),
                CustomerId = reader.GetInt32(1),
                Amount = reader.GetDecimal(2),
                OrderDate = reader.GetDateTime(3),
                Status = reader.GetString(4)
            });
        }

        return Ok(orders);
    }
}

Shard Router

The router contains the logic that determines where the customer data lives.

public interface IShardRouter
{
    string GetConnectionString(long customerId);
}

public class ShardRouter : IShardRouter
{
    private readonly IConfiguration _configuration;

    public ShardRouter(IConfiguration configuration)
    {
        _configuration = configuration;
    }

    public string GetConnectionString(long customerId)
    {
        if (customerId <= 1_000_000)
        {
            return _configuration.GetConnectionString("Shard1")!;
        }

        if (customerId <= 2_000_000)
        {
            return _configuration.GetConnectionString("Shard2")!;
        }

        return _configuration.GetConnectionString("Shard3")!;
    }
}

Connection Strings

appsettings.json can contain the shard connections:

{
  "ConnectionStrings": {
    "Shard1": "Server=sql01;Database=OrdersShard1;...",
    "Shard2": "Server=sql02;Database=OrdersShard2;...",
    "Shard3": "Server=sql03;Database=OrdersShard3;..."
  }
}

In production, credentials should be stored securely rather than committed directly into configuration files.

Register the Router

builder.Services.AddScoped<IShardRouter, ShardRouter>();

builder.Services.AddControllers();

var app = builder.Build();

app.MapControllers();

app.Run();

Request Example

Request:

GET /api/customers/1500000/orders

The routing sequence becomes:

customerId = 1,500,000
        ↓
ShardRouter
        ↓
Shard 2
        ↓
OrdersShard2
        ↓
SELECT ... WHERE CustomerId = 1500000

Expected response:

[
  {
    "orderId": 3,
    "customerId": 1500000,
    "amount": 3000.00,
    "orderDate": "2026-09-03",
    "status": "Completed"
  },
  {
    "orderId": 4,
    "customerId": 1800000,
    "amount": 4500.00,
    "orderDate": "2026-09-04",
    "status": "Pending"
  }
]

For the exact customer query, only the database containing that customer's data should be accessed.

Important: Don't Use HTTP Routing Alone

The ASP.NET Core route identifies the customer, but the shard routing logic decides the database.

HTTP Route
    ↓
Controller
    ↓
Shard Router
    ↓
Connection String
    ↓
SQL Server Shard
    ↓
Query

Shard Mapping Table

Instead of hardcoding ranges, production systems commonly maintain a shard map.

ShardId    CustomerStart    CustomerEnd
-------    -------------    -----------
1          1                1000000
2          1000001          2000000
3          2000001          3000000

The router can then look up the correct shard dynamically.

flowchart LR
    A[API Request] --> B[CustomerId]
    B --> C[Shard Mapping]
    C --> D[Shard Connection]
    D --> E[SQL Server]

Advantages

Distributes database storage.
Distributes database traffic.
Allows independent scaling of shards.
One overloaded shard does not necessarily overload every shard.
Can support very large datasets.

Challenges

More operational complexity.
Cross-shard queries are difficult.
Cross-shard transactions are expensive and complex.
Rebalancing shards can be difficult.
Shard key selection becomes critical.
Monitoring and backup become more complex.

Interview Scenario

Interviewer: Your SQL Server database has reached several terabytes and traffic continues to increase. How would you scale it?

Answer: I would first determine whether indexing, query optimization, read replicas, partitioning, and vertical scaling are sufficient. If the workload still exceeds the capacity of one database, I would consider sharding. For example, I could shard by CustomerId. The ASP.NET Core API would use a shard router to determine the appropriate database before executing the query.

Common Interview Questions

What is database sharding?
Why do we need sharding?
Sharding vs partitioning?
Where should shard-routing logic live?
How does an ASP.NET Core API select a shard?
What is a shard key?
What happens to cross-shard queries?
How do you handle transactions across shards?
How do you rebalance shards?
What happens if one shard goes down?

Key Takeaways

Sharding distributes data across multiple databases.
The application or a routing layer determines the target shard.
CustomerId is a common example of a shard key.
ASP.NET Core route parameters can provide the key needed by the shard router.
Sharding improves horizontal database scalability.
Cross-shard operations introduce significant complexity.
Always evaluate partitioning, replicas, indexing, and query optimization before introducing sharding.

Articles categories

Projects categories

Cheatsheets categories

Videos categories

eBooks categories

Latest articles