Database Sharding Manager
Overview
Implement and manage horizontal database sharding strategies across PostgreSQL, MySQL, and MongoDB. This skill covers shard key selection, data distribution analysis, cross-shard query routing, and rebalancing operations for databases that have outgrown single-node capacity.
Prerequisites
- Database admin credentials with CREATE DATABASE, CREATE TABLE, and replication permissions
psql, mysql, or mongosh CLI tools installed and configured- Network connectivity between all shard nodes
- Current table sizes and growth rate data (query
pg_total_relation_size or information_schema.TABLES)
- Application query patterns documented or access to slow query logs
- Enough disk and memory on target shard nodes to handle redistributed data
Instructions
Analyze the current database size and identify tables exceeding single-node capacity thresholds (typically >500GB or >1B rows). Run SELECT pg_size_pretty(pg_total_relation_size('table_name')) for PostgreSQL or SELECT data_length + index_length FROM information_schema.TABLES for MySQL.
Evaluate candidate shard keys by examining query WHERE clauses, JOIN patterns, and data distribution. A good shard key has high cardinality, even distribution, and appears in most queries. Run SELECT shard_key_column, COUNT(*) FROM table GROUP BY shard_key_column ORDER BY COUNT(*) DESC LIMIT 20 to check distribution.
Choose a sharding strategy based on workload patterns:
- Hash-based: Even distribution, best for key-value lookups. Use
hash(shard_key) % num_shards.
- Range-based: Good for time-series or sequential data. Partition by date ranges or ID ranges.
- Directory-based: Maximum flexibility with a lookup table mapping keys to shards.
- Geographic: Route by region for data residency or latency requirements.
Design the shard topology by determining the number of shards, replication factor, and placement. For PostgreSQL, use Citus extension or manual foreign data wrappers. For MySQL, configure vitess or ProxySQL routing. For MongoDB, enable sharding on the cluster with sh.enableSharding() and sh.shardCollection().
Create the shard schema on all target nodes, ensuring identical table definitions, indexes, and constraints across every shard. Generate DDL scripts and verify with checksums.
Implement the routing layer that directs queries to the correct shard. This can be application-level (connection selection based on shard key), middleware (ProxySQL, PgBouncer with routing), or database-native (Citus, MongoDB mongos).
Migrate existing data to shards using batch operations. Extract data in chunks of 10,000-50,000 rows, transform shard key assignments, and load into target shards. Verify row counts match after migration.
Validate cross-shard queries work correctly, especially aggregations and JOINs that span multiple shards. Test scatter-gather query performance and implement application-level aggregation where needed.
Set up monitoring for shard balance (data size per shard, query load per shard) and configure alerts for skew exceeding 20% deviation from the average.
Document the shard map, routing logic, and rebalancing procedures for operational runbooks.
Output
- Shard key analysis report with cardinality, distribution histograms, and recommended key selection
- Shard topology diagram mapping databases, tables, and key ranges to physical nodes
- DDL migration scripts for creating shard schemas with matching indexes and constraints
- Routing configuration files for ProxySQL, Citus, vitess, or application-level routing
- Data migration scripts with batch extraction, transformation, and verification queries
- Monitoring queries for shard balance, cross-shard query latency, and hotspot detection
Error Handling
| Error |
Cause |
Solution |
| Hotspot shard receiving disproportionate traffic |
Poor shard key choice with low cardinality or skewed distribution |
Re-analyze shard key distribution; consider compound shard keys or hash-based sharding |
| Cross-shard JOIN timeout |
Scatter-gather query across too many shards |
Denormalize frequently joined data onto the same shard; use application-level aggregation |
| Shard rebalancing data loss |
Migration interrupted mid-batch without transaction wrapping |
Wrap batch migrations in transactions; verify source and destination row counts before deleting source data |
| Connection pool exhaustion |
Each shard requires its own connection pool, multiplying total connections |
Reduce per-shard pool size; use connection multiplexing with PgBouncer or ProxySQL |
| Schema drift between shards |
DDL changes applied to some shards but not others |
Use centralized DDL deployment scripts; verify schema checksums across all shards after changes |
Examples
E-commerce order table sharding by customer_id: A 2TB orders table with 800M rows is sharded across 8 nodes using hash-based distribution on customer_id. All queries for a single customer hit one shard. Cross-customer analytics queries use a separate read replica with full data.
Time-series IoT data with range sharding: Sensor readings partitioned by month into separate shards. Each shard holds one month of data. Queries for recent data hit the active shard; historical analysis queries span multiple shards with parallel execution. Old shards are archived to cold storage quarterly.
Multi-tenant SaaS with directory-based sharding: A tenant-to-shard lookup table routes each tenant to a dedicated shard. Large tenants get dedicated shards; small tenants share shards. Rebalancing moves tenants between shards by updating the directory and migrating data.
Resources
1---2name: managing-database-sharding-23description: Process use when you need to work with database sharding. This skill provides horizontal sharding strategies with comprehensive guidance and automation. Trigger with phrases like "implement sharding", "shard database", or "distribute data".4license: MIT5---6# Database Sharding Manager
7
8## Overview
9
10Implement and manage horizontal database sharding strategies across PostgreSQL, MySQL, and MongoDB. This skill covers shard key selection, data distribution analysis, cross-shard query routing, and rebalancing operations for databases that have outgrown single-node capacity.
11
12## Prerequisites
13
14- Database admin credentials with CREATE DATABASE, CREATE TABLE, and replication permissions
15- `psql`, `mysql`, or `mongosh` CLI tools installed and configured
16- Network connectivity between all shard nodes
17- Current table sizes and growth rate data (query `pg_total_relation_size` or `information_schema.TABLES`)
18- Application query patterns documented or access to slow query logs
19- Enough disk and memory on target shard nodes to handle redistributed data
20
21## Instructions
22
231. Analyze the current database size and identify tables exceeding single-node capacity thresholds (typically >500GB or >1B rows). Run `SELECT pg_size_pretty(pg_total_relation_size('table_name'))` for PostgreSQL or `SELECT data_length + index_length FROM information_schema.TABLES` for MySQL.
24
252. Evaluate candidate shard keys by examining query WHERE clauses, JOIN patterns, and data distribution. A good shard key has high cardinality, even distribution, and appears in most queries. Run `SELECT shard_key_column, COUNT(*) FROM table GROUP BY shard_key_column ORDER BY COUNT(*) DESC LIMIT 20` to check distribution.
26
273. Choose a sharding strategy based on workload patterns:
28 - **Hash-based**: Even distribution, best for key-value lookups. Use `hash(shard_key) % num_shards`.
29 - **Range-based**: Good for time-series or sequential data. Partition by date ranges or ID ranges.
30 - **Directory-based**: Maximum flexibility with a lookup table mapping keys to shards.
31 - **Geographic**: Route by region for data residency or latency requirements.
32
334. Design the shard topology by determining the number of shards, replication factor, and placement. For PostgreSQL, use Citus extension or manual foreign data wrappers. For MySQL, configure vitess or ProxySQL routing. For MongoDB, enable sharding on the cluster with `sh.enableSharding()` and `sh.shardCollection()`.
34
355. Create the shard schema on all target nodes, ensuring identical table definitions, indexes, and constraints across every shard. Generate DDL scripts and verify with checksums.
36
376. Implement the routing layer that directs queries to the correct shard. This can be application-level (connection selection based on shard key), middleware (ProxySQL, PgBouncer with routing), or database-native (Citus, MongoDB mongos).
38
397. Migrate existing data to shards using batch operations. Extract data in chunks of 10,000-50,000 rows, transform shard key assignments, and load into target shards. Verify row counts match after migration.
40
418. Validate cross-shard queries work correctly, especially aggregations and JOINs that span multiple shards. Test scatter-gather query performance and implement application-level aggregation where needed.
42
439. Set up monitoring for shard balance (data size per shard, query load per shard) and configure alerts for skew exceeding 20% deviation from the average.
44
4510. Document the shard map, routing logic, and rebalancing procedures for operational runbooks.
46
47## Output
48
49- **Shard key analysis report** with cardinality, distribution histograms, and recommended key selection
50- **Shard topology diagram** mapping databases, tables, and key ranges to physical nodes
51- **DDL migration scripts** for creating shard schemas with matching indexes and constraints
52- **Routing configuration** files for ProxySQL, Citus, vitess, or application-level routing
53- **Data migration scripts** with batch extraction, transformation, and verification queries
54- **Monitoring queries** for shard balance, cross-shard query latency, and hotspot detection
55
56## Error Handling
57
58| Error | Cause | Solution |
59|-------|-------|---------|
60| Hotspot shard receiving disproportionate traffic | Poor shard key choice with low cardinality or skewed distribution | Re-analyze shard key distribution; consider compound shard keys or hash-based sharding |
61| Cross-shard JOIN timeout | Scatter-gather query across too many shards | Denormalize frequently joined data onto the same shard; use application-level aggregation |
62| Shard rebalancing data loss | Migration interrupted mid-batch without transaction wrapping | Wrap batch migrations in transactions; verify source and destination row counts before deleting source data |
63| Connection pool exhaustion | Each shard requires its own connection pool, multiplying total connections | Reduce per-shard pool size; use connection multiplexing with PgBouncer or ProxySQL |
64| Schema drift between shards | DDL changes applied to some shards but not others | Use centralized DDL deployment scripts; verify schema checksums across all shards after changes |
65
66## Examples
67
68**E-commerce order table sharding by customer_id**: A 2TB orders table with 800M rows is sharded across 8 nodes using hash-based distribution on `customer_id`. All queries for a single customer hit one shard. Cross-customer analytics queries use a separate read replica with full data.
69
70**Time-series IoT data with range sharding**: Sensor readings partitioned by month into separate shards. Each shard holds one month of data. Queries for recent data hit the active shard; historical analysis queries span multiple shards with parallel execution. Old shards are archived to cold storage quarterly.
71
72**Multi-tenant SaaS with directory-based sharding**: A tenant-to-shard lookup table routes each tenant to a dedicated shard. Large tenants get dedicated shards; small tenants share shards. Rebalancing moves tenants between shards by updating the directory and migrating data.
73
74## Resources
75
76- PostgreSQL Citus documentation: https://docs.citusdata.com/
77- MySQL Vitess documentation: https://vitess.io/docs/
78- MongoDB Sharding guide: https://www.mongodb.com/docs/manual/sharding/
79- ProxySQL query routing: https://proxysql.com/documentation/
80- Shard key design patterns: https://www.mongodb.com/docs/manual/core/sharding-choose-a-shard-key/
jeremylongshore