Sharding, Autosharding & Autoscaling

Distributed Systems intermediate Building with AI 11 min read

Overview

Sharding is the practice of horizontally partitioning data across multiple database instances. Instead of storing an entire dataset on a single server, sharding splits it into smaller pieces called shards, each living on a separate node. Every shard contains the same columns but different rows, and a routing layer directs queries to the correct shard based on a shard key.

For AI engineering, sharding is critical because vector databases, document stores, and embedding indexes can grow to billions of records. A single-node architecture cannot handle the write throughput of continuous embedding ingestion or the read latency requirements of real-time semantic search. Systems like Weaviate, Milvus, and Qdrant all rely on sharding to distribute vector data across a cluster.

Autosharding and autoscaling take this further by removing manual intervention. Instead of a human deciding when and how to split data, the system monitors shard sizes, query hotspots, and resource utilization, then automatically rebalances. This is the difference between a system that works at prototype scale and one that operates in production under unpredictable load.

How It Works

Shard Key Selection

The shard key is the field (or combination of fields) used to determine which shard a record belongs to. Choosing the right shard key is the single most important sharding decision because it determines data distribution, query routing efficiency, and whether hotspots will form.

Key Concept: Shard Key

A shard key must have high cardinality (many distinct values) to distribute data evenly, and it should align with your most common query patterns so that most queries hit a single shard rather than scattering across all of them.

Sharding Strategies

Hash-Based Sharding: A hash function is applied to the shard key, and the hash value determines the target shard. For example, shard_id = hash(document_id) % num_shards. This produces an even distribution of data across shards, eliminating hotspots. The tradeoff is that range queries (e.g., “all documents from January”) become scatter-gather operations because adjacent key values land on different shards.

// Hash-based shard assignment
func getShardID(documentID string, numShards int) int {
    h := murmur3.Sum64([]byte(documentID))
    return int(h % uint64(numShards))
}

Range-Based Sharding: Records are assigned to shards based on value ranges of the shard key. For example, shard 0 holds IDs 1-1,000,000, shard 1 holds 1,000,001-2,000,000, and so on. Range sharding keeps related data together, making range scans efficient. The downside is that new data often clusters at one end of the range (e.g., the newest timestamps), creating write hotspots on a single shard.

Geographic (Geo) Sharding: The shard key is derived from geographical location. Data for European users lives on EU servers, North American data on US servers, and so on. This reduces latency by keeping data physically close to users and can help meet data residency regulations like GDPR. The challenge is handling users who move between regions or queries that span multiple geographies.

Directory-Based Sharding: A lookup table maps each key to its shard. This is the most flexible approach—you can move individual records between shards without changing the key—but the directory itself becomes a single point of failure and a potential bottleneck.

How Vector Databases Shard

Vector databases like Weaviate use a virtual shard system. A 64-bit Murmur-3 hash of each object’s UUID determines placement across virtual shards, which are then mapped to physical nodes. This two-level indirection makes rebalancing cheaper: moving a virtual shard to a new physical node requires updating only the mapping table, not rehashing every record.

Milvus takes a different approach, sharding by collection and using a message queue (Pulsar or Kafka) to buffer writes. Each shard corresponds to a channel in the message queue, and data nodes subscribe to channels to ingest their assigned data.

Autosharding: Automatic Split and Merge

Manual sharding requires operators to predict data growth and manually split shards when they get too large. Autosharding automates this process:

Key Concept: Databricks Dicer

Dicer is Databricks’ open-source auto-sharder that dynamically manages shard assignments. It handles restarts, failures, and shifting workloads by continuously monitoring shard health and rebalancing assignments. It prevents load imbalance from hot keys and allows seamless autoscaling without loss of cache hit rate.

Autoscaling

Autoscaling adds or removes compute nodes based on load metrics. In a sharded system, autoscaling and autosharding work together:

  1. Monitoring detects that query latency or CPU utilization has exceeded a threshold.
  2. The autoscaler provisions new nodes and adds them to the cluster.
  3. The autosharder rebalances existing shards onto the new nodes, spreading load.
  4. When load decreases, the autoscaler drains nodes by migrating their shards elsewhere, then removes them.

Cloud-native databases like YugabyteDB, CockroachDB, and TiDB support this natively. For vector databases, Weaviate and Milvus both support horizontal autoscaling in their managed cloud offerings, automatically adding query nodes when search latency degrades.

Practical Applications

Embedding Storage at Scale

A production RAG system ingesting millions of documents generates billions of embedding vectors. At 1536 dimensions (OpenAI text-embedding-3-large) with 4-byte floats, each vector is ~6 KB. One billion vectors requires ~6 TB of raw vector data alone, before indexes. This data must be sharded across multiple nodes, with each shard maintaining its own HNSW index for fast approximate nearest-neighbor search.

Multi-Tenant AI Platforms

SaaS AI platforms serving multiple customers often shard by tenant ID. Each tenant’s embeddings, documents, and metadata live on a dedicated shard (or set of shards), providing data isolation and preventing one tenant’s heavy usage from degrading another’s performance. Autoscaling ensures that when a large tenant onboards a massive document corpus, new nodes are provisioned automatically.

Global Deployment

A hospitality AI platform serving properties worldwide might geo-shard its knowledge base: European hotel data on EU nodes, Asian properties on APAC nodes. This reduces search latency for local queries and satisfies data residency requirements. Cross-region queries (e.g., corporate analytics) use scatter-gather across all shards.

Key Takeaways

Further Reading