Sharding-JDBC Deep Dive: Scaling to 100M Orders, Fixing Shard Skew & Cross-DB Joins
This article shares real-world experience implementing database sharding with Sharding-JDBC for an e-commerce platform, covering sharding key selection, distributed ID generation, cross-database join avoidance, pagination fixes, zero-downtime migration, and shard skew remediation.
When to Shard: Exhaust Simpler Optimizations First
The team behind a major e-commerce platform (referred to as "Mall") faced an order table exceeding 12 million rows, with database CPU regularly hitting 85%+ during promotions. Before resorting to sharding, they applied four preliminary optimizations:
Index optimization and SQL rewriting to eliminate slow queries.
Read-write separation to offload read traffic to replicas.
Cold-hot data separation: archiving orders older than one year to an archive database.
Vertical splitting: moving large fields like order remarks and logistics details into a separate order-extension table.
Only when these measures failed to handle peak loads and the business projected 100M+ rows within 1-2 years did they proceed with sharding. The article stresses: sharding is an invasive architectural change; avoid it unless absolutely necessary.
Middleware Selection: Sharding-JDBC vs Sharding-Proxy
Because the stack is Java microservices with no dedicated DBA team, they chose Sharding-JDBC (client-side library) over Sharding-Proxy (standalone proxy). Key trade-offs:
Sharding-JDBC : No extra proxy service, lower latency, simple deployment; but couples to Java and invades application code.
Sharding-Proxy : Language-agnostic, centralized governance; but adds network hop and requires proxy cluster maintenance.
For most Java microservice teams, Sharding-JDBC is the pragmatic default.
Sharding Strategy: Choosing the Shard Key
The shard key determines data distribution and query routing. Two main approaches were evaluated:
1. Hash Sharding (Chosen)
Using userId as the shard key with a hash-mod algorithm. Rationale: the vast majority of order queries ("my orders", order detail) carry the user ID, enabling single-shard routing.
spring: shardingsphere: rules: sharding: tables: t_order: actual-data-nodes: ds$->{0..3}.t_order_$->{0..3} database-strategy: standard: sharding-column: user_id sharding-algorithm-name: order-hash-mod sharding-algorithms: order-hash-mod: type: HASH_MOD props: sharding-count: 4Pros : Uniform data distribution, balanced load.
Cons : Queries without userId trigger broadcast across all shards, hurting performance.
2. Range Sharding
Based on time or auto-increment ID intervals. Friendly for time-range queries and easy scaling, but creates hotspot shards where new data concentrates. Suitable for logs or flow tables.
Production lesson : Pick a high-frequency query field as shard key. If most queries lack the shard key, broadcast queries will overwhelm the database.
Shard Skew: The Hidden Trap
Early on, a few large B2B merchants generated hundreds of thousands of orders each. Hashing their userId sent all those rows to a single shard, causing severe skew, index degradation, and slow SQL. Fixes applied:
Composite sharding algorithm or application-level re-sharding for identified giant merchants.
Historical data re-sharded to rebalance.
Monitoring alerts on per-shard row counts to catch skew early.
Production Pain Points & Solutions
4.1 Distributed ID Generation
Auto-increment IDs break across shards. Sharding-JDBC provides segment mode and Snowflake. The team chose segment mode for higher throughput and to avoid Snowflake's clock-drift risk.
spring: shardingsphere: rules: sharding: tables: t_order: key-generate-strategy: column: order_id key-generator-name: order-snowflake key-generators: order-snowflake: type: SNOWFLAKE4.2 Cross-Database Joins
After sharding, joining orders with user or membership tables across databases is impossible. Three practical workarounds:
In-memory join in business layer : Fetch both sides separately and merge in Java. Only for small result sets.
Field redundancy (heavily used) : Duplicate user nickname and membership level into t_order. Trade storage for query speed; most order queries no longer need the user database.
Wide-table sync to Elasticsearch : Complex multi-condition order statistics are synced via binlog to ES; the database handles only basic CRUD.
Warning : Don't rely on middleware to magically solve cross-shard joins; it's an architectural design issue best avoided at the business level.
4.3 Pagination Pitfalls
Using LIMIT OFFSET directly on sharded tables returns wrong results because each shard paginates locally. Solutions:
In-memory merge pagination (fetch top N from each shard, merge, then paginate).
For deep pagination, abandon offset; use cursor-based pagination with order_id as a scroll pointer.
// Cursor pagination pseudo-code for mall admin order query// No LIMIT OFFSET; pass last page's max orderIdList<Order> list = orderMapper.queryByUserIdAndMinOrderId(userId, lastMaxOrderId, pageSize);4.4 Distributed Transactions
Cross-shard updates break local transactions. Primary strategy: constrain all operations for a single user to the same shard (via userId sharding) to avoid distributed transactions entirely. When unavoidable, use Seata AT mode with acceptance of performance overhead and proper compensation logic.
Zero-Downtime Data Migration: Dual-Write Cutover
The live system could not stop. They followed the industry-standard dual-write migration:
Build new sharded cluster : Provision new databases, configure Sharding-JDBC rules; old cluster stays live.
Enable dual-write : All inserts, updates, deletes go to both old and new clusters.
Full historical sync : Batch-copy tens of millions of existing orders to new shards with idempotent writes to prevent duplicates.
Data validation : Compare row counts, order amounts, status fields between old and new; fix mismatches.
Canary cutover : Shift 10% of user traffic to new cluster; monitor errors, slow queries, metrics.
Full switch : Move 100% traffic; keep old cluster as read-only backup for 30 days before decommission.
Critical : A complete rollback plan must exist to instantly revert to the old cluster if the new one misbehaves.
Incident Postmortem: Shard Skew Causes Slow Queries
Two months post-launch, order query latency spiked and CPU occasionally surged. Investigation steps:
Monitoring revealed one shard held disproportionately more data.
Root cause: top-tier merchants with hundreds of thousands of orders all hashed to the same shard.
That shard's table grew huge, indexes became inefficient, generating slow SQL that dragged down the whole app.
Remediation:
Application-level detection of giant merchants with custom re-sharding logic.
Historical data re-sharded to redistribute load evenly.
Added per-shard row-count monitoring with alerts to catch skew early.
Takeaway : Sharding launch is not the end; continuous monitoring of shard distribution is essential because skew is a chronic, worsening problem.
Architect's Summary
Sharding is high-complexity; exhaust indexing, archiving, read-write separation first. The mall only sharded after all else failed.
Shard key choice is make-or-break; prioritize high-frequency query fields to avoid broadcast queries.
Cross-shard joins, pagination, distributed IDs, and distributed transactions are frequent pitfalls; design solutions upfront, don't expect middleware to auto-resolve.
Migration must use dual-write + canary + full rollback; never do stop-the-world migration.
Post-launch, monitor shard data distribution continuously to detect skew early.
Final thought : Sharding tests not coding skill but upfront business assessment, solution design, and migration planning. Get the design right and you avoid 80% of the pitfalls.
Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
Internet Backend Architecture Stack
Over ten years of hands-on Java backend architecture experience, focusing on sharing mainstream internet technologies such as Java backend, microservices, distributed systems, high concurrency, performance optimization, and source code analysis. Building a platform for Java technologists to exchange and grow, collaborating with peers to advance together.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.
