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
Converted and distributed by TomeVault — claim your Tome and manage your conversions.
1---2name: jeremylongshore-claude-code-plugins-plus-skills-managing-33description: Database Sharding Manager4---5# Database Sharding Manager67## Overview89Implement 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.1011## Prerequisites1213- Database admin credentials with CREATE DATABASE, CREATE TABLE, and replication permissions14- `psql`, `mysql`, or `mongosh` CLI tools installed and configured15- Network connectivity between all shard nodes16- Current table sizes and growth rate data (query `pg_total_relation_size` or `information_schema.TABLES`)17- Application query patterns documented or access to slow query logs18- Enough disk and memory on target shard nodes to handle redistributed data1920## Instructions21221. 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.23242. 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.25263. Choose a sharding strategy based on workload patterns:27 - **Hash-based**: Even distribution, best for key-value lookups. Use `hash(shard_key) % num_shards`.28 - **Range-based**: Good for time-series or sequential data. Partition by date ranges or ID ranges.29 - **Directory-based**: Maximum flexibility with a lookup table mapping keys to shards.30 - **Geographic**: Route by region for data residency or latency requirements.31324. 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()`.33345. 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.35366. 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).37387. 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.39408. 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.41429. 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.434410. Document the shard map, routing logic, and rebalancing procedures for operational runbooks.4546## Output4748- **Shard key analysis report** with cardinality, distribution histograms, and recommended key selection49- **Shard topology diagram** mapping databases, tables, and key ranges to physical nodes50- **DDL migration scripts** for creating shard schemas with matching indexes and constraints51- **Routing configuration** files for ProxySQL, Citus, vitess, or application-level routing52- **Data migration scripts** with batch extraction, transformation, and verification queries53- **Monitoring queries** for shard balance, cross-shard query latency, and hotspot detection5455## Error Handling5657| Error | Cause | Solution |58|-------|-------|---------|59| 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 |60| Cross-shard JOIN timeout | Scatter-gather query across too many shards | Denormalize frequently joined data onto the same shard; use application-level aggregation |61| 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 |62| 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 |63| 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 |6465## Examples6667**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.6869**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.7071**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.7273## Resources7475- PostgreSQL Citus documentation: https://docs.citusdata.com/76- MySQL Vitess documentation: https://vitess.io/docs/77- MongoDB Sharding guide: https://www.mongodb.com/docs/manual/sharding/78- ProxySQL query routing: https://proxysql.com/documentation/79- Shard key design patterns: https://www.mongodb.com/docs/manual/core/sharding-choose-a-shard-key/8081---82> Converted and distributed by [TomeVault](https://tomevault.io/claim/jeremylongshore) — claim your Tome and manage your conversions.83<!-- tomevault:4.0:skill_md:2026-04-11 -->
tomevault-io