A well-tuned PostgreSQL instance can handle roughly 10,000 writes per second. That sounds like a lot — until you consider that [Uber](/learn/system-design/questions/uber) processes millions of driver location updates per minute, or that the [Ad Click Aggregator](/learn/system-design/questions/adds-click-aggregator) ingests millions of click events per second during a Super Bowl ad campaign. When reads are the bottleneck, you can throw caches and replicas at the problem (see [Scaling Reads](/learn/system-design/design-patterns/scaling-reads)). But writes are harder — every write must reach the source of truth, and there's no "write cache" that magically solves it. The good news: there's a clear toolkit for scaling writes, and each tool addresses a different part of the problem. A combination of four strategies will allow you to scale writes beyond what a single, unoptimized database or server can handle: vertical scaling and database choices, sharding and partitioning, handling bursts with queues and load shedding, and batching with hierarchical aggregation. ## Tier 1 — Squeeze More From Your Existing Setup Before splitting your data across machines, make sure you've exhausted the hardware at your disposal. We want to show our interviewer that we've done our due diligence and we're not prematurely adding complexity when hardware or local software tweaks will do it. ### Vertical Scaling Writes are bottlenecked by disk I/O, CPU, or network bandwidth. Before re-architecting, confirm you're actually hitting those walls. Often this means doing some quick back-of-the-envelope math to check (a) what your write throughput actually is, and (b) whether that fits within your hardware capabilities. Many candidates assume instances with 4-8 cores and a single spinning-platter hard disk. But cloud providers routinely offer systems with 200+ CPU cores and 10-gigabit network interfaces. Hardware can go further than you think — it's worth brushing up on your numbers to make sure you're not missing potential. You can make the case (and most interviewers will appreciate it) that some of your challenge is solved with modern hardware. However, many interviewers have an assessment built around non-vertical scaling, and they'll frequently move the goalposts until you're forced to contend with scale in other ways. So make the case, but try not to get into a back-and-forth if the interviewer isn't receptive. ### Database Choices and Write-Optimized Storage Most general-purpose databases are designed for mixed workloads — reads and writes. When you know your system is write-heavy, you can make targeted choices that dramatically improve performance. Databases like Cassandra, RocksDB, and ScyllaDB use **LSM-trees** (Log-Structured Merge Trees) instead of B-Trees. The difference in write performance is dramatic: 1. The write goes to an **in-memory memtable** (sorted buffer) — this is nearly instant 2. Simultaneously, it's appended to a **write-ahead log** (WAL) for durability — sequential I/O, very fast 3. When the memtable fills up, it's flushed to disk as an immutable **SSTable** (Sorted String Table) 4. Background **compaction** merges SSTables to keep read performance reasonable The trick: writes never update data in place on disk. They're always either in memory or appended sequentially. Sequential I/O is orders of magnitude faster than random I/O on both SSDs and spinning disks. **The trade-off:** reads are slower because data might be spread across multiple SSTables. LSM-trees trade read performance for write performance — which is exactly what you want in a write-heavy system. Other databases make similar trade-offs: Beyond choosing the right database, you can optimize any database for writes: disable expensive features like foreign key constraints or complex triggers during high-write periods, tune write-ahead logging to batch multiple transactions before flushing to disk, and reduce index overhead (fewer indexes mean faster writes, though you'll pay for it on reads). In interviews, mentioning specific database choices shows you understand the trade-offs. Don't just say "use a faster database" — explain why Cassandra's append-only writes are faster than MySQL's B-tree updates, or why you might choose a time-series database for metrics collection. **See it in practice:** In the [Ad Click Aggregator](/learn/system-design/questions/adds-click-aggregator), Cassandra (LSM-tree) stores raw click events because the write rate is too high for a B-Tree database. In the [Leaderboard](/learn/system-design/questions/leaderboard), Redis sorted sets act as an in-memory write-optimized store for rapid score updates. The [Time-Series & Metrics Ingestion](/learn/system-design/design-patterns/time-series-metrics-ingestion) pattern covers how time-series databases use LSM-tree variants optimized for append-heavy workloads. ## Tier 2 — Split Data Across Multiple Machines When one well-tuned database isn't enough, you need to distribute writes across multiple nodes. If one server can handle 10,000 writes/second, then 10 servers should handle 100,000 writes/second. The key word is "should" — it depends on how well you distribute the load. ### Sharding (Horizontal Partitioning) **Sharding** means splitting your data across multiple independent database instances, each responsible for a disjoint subset. If one database handles 10K writes/s, four shards handle 40K, and you can keep adding more for roughly linear scaling. The most important decision is your **partition key** — the column that determines which shard a row belongs to. There's no universally correct choice; it depends on your access patterns: **The golden rule:** choose a key that distributes writes evenly AND keeps related data together. User ID is the most common choice because a single user's data rarely needs to be joined with another user's data. Keep in mind that you also need to consider how the data will be read. If you spread all writes across shards but each request needs to collect data from every single shard, you'll have significant overhead. Ask yourself "how many shards does this request need to hit?" and "how often does this request happen?" Those questions will help you decide whether you've created a bottleneck. ### Consistent Hashing When you add or remove shards, you need to move data around. With naive modulo hashing (`hash(key) % N`), changing N from 3 to 4 remaps roughly 75% of all keys — a massive, disruptive data migration. **Consistent hashing** arranges nodes on a virtual ring. Each key maps to the next node clockwise on the ring. When you add a node, only the keys between it and its predecessor need to move — roughly `K/N` keys instead of almost all of them. This makes scaling out (or handling node failures) far less painful. In practice, each physical node gets multiple **virtual nodes** on the ring to ensure even distribution. Without virtual nodes, you'd get unlucky clusters where one node owns a disproportionate range. **See it in practice:** In the [Distributed Cache](/learn/system-design/questions/cache), consistent hashing determines which cache node stores each key, and virtual nodes prevent uneven load. In [Uber](/learn/system-design/questions/uber), driver location data is sharded geographically using consistent hashing to distribute writes across Redis GEO nodes. The [Distributed Caching](/learn/system-design/design-patterns/distributed-caching) pattern covers consistent hashing in depth alongside replication and failover. ### Vertical Partitioning While horizontal sharding splits rows, **vertical partitioning** splits columns. You separate different types of data that have different access patterns and scaling requirements. Instead of cramming everything into one massive table, you break it apart based on how the data is actually used. Think of a social media post. In a monolithic approach, you might have a single table with the post content, engagement metrics, and analytics data all crammed together. This table gets hammered from all directions — users write content, the system updates engagement metrics constantly, and analytics queries scan through massive amounts of data. With vertical partitioning, you split this into specialized tables. Core post content is write-once, read-many — it goes in a table optimized with traditional B-tree indexes. Engagement metrics (likes, comments, views) see high-frequency updates — they might use in-memory storage or specialized counters. Analytics data is append-only — it can use time-series optimized storage with column-oriented compression. Once you've logically separated your data, you can also move each table to a different database instance optimized for its specific access pattern. The data modeling challenge is as much about how you logically think about your data as it is about where it physically lives. **See it in practice:** In [Meta Post Privacy](/learn/system-design/questions/facebook-post-privacy), separating post content from privacy settings and engagement metrics allows each to be scaled independently. In [Slack](/learn/system-design/questions/slack), message content, channel metadata, and user presence are vertically partitioned for independent scaling. The [Event Sourcing & CQRS](/learn/system-design/design-patterns/event-sourcing-cqrs) pattern takes vertical partitioning further by maintaining completely separate read and write models. ## Tier 3 — Handle Bursts and Reduce Write Volume While partitioning and sharding will get you most of the way to scale, real-world write traffic isn't steady. Interviewers love to drill in on things like "what happens on Black Friday, when order volume 4x's?" or "during New Year's, we triple the number of drivers on the road." If we need to handle 4x write volume at peak, we can only be using 25% of capacity during quieter times. Most systems aren't scaled this way. This means we either need to buffer the writes or get rid of some of them in a way that's acceptable to the business. ### Write Queues and Buffering What happens when traffic spikes 10x during peak hours? Without a buffer, those writes slam directly into the database, overwhelming it. A **write queue** (Kafka, Amazon Kinesis, or Redis Streams) sits between your application and the database, absorbing bursts and delivering a smooth, steady stream of writes. The key insight: the queue **decouples the write rate your app produces from the write rate your database can consume**. During a spike, the queue grows. Once the spike passes, the consumers catch up. The database never sees more than it can handle. Because queues are inherently async, the app server only knows the write was recorded in the queue, not that it was successfully written to the database. In most cases, clients will need a way to check the write was eventually committed. But queues are only a temporary solution. If the app server continues to write faster than records can be committed to the database, you get unbounded queue growth. Writes take longer and longer to process. Use queues when you expect bursts that are short-lived, not to patch a database that can't handle the steady-state load. **See it in practice:** In the [Webhook](/learn/system-design/questions/webhook), delivery results are queued before being written to the database so that a burst of webhook callbacks doesn't exhaust the DB. In [Slack](/learn/system-design/questions/slack), message writes are buffered through a queue to handle spikes when thousands of users send messages simultaneously. The [Pub/Sub Messaging](/learn/system-design/design-patterns/pub-sub-messaging) pattern goes deeper into Kafka topics, partitions, and consumer groups. ### Load Shedding When your system is overwhelmed, sometimes the best move is to intentionally drop writes. This sounds like a cop-out, but it's actually a powerful tool. **Load shedding** tries to determine which writes are most important to the business and drops the rest. Consider problems like [Uber](/learn/system-design/questions/uber) where drivers report their locations every few seconds. If we have too many drivers, adding a queue just delays the inevitable — a blown-out queue. But if we take a step back, we realize drivers are going to send a fresh update in a few seconds anyway. If we drop one write, a fresher one is coming right behind it. A simple approach: drop the least useful writes during system overload. For ride-sharing, these might be location updates within seconds of a previous update. For an analytics system, you might drop impressions to ensure you can process the more important click events. Putting release valves in place shows you can keep a bad situation (too much load) from turning into a disaster (system failure), even if it means a slightly degraded experience for some users. **See it in practice:** In [Uber](/learn/system-design/questions/uber), driver location updates can tolerate dropping during extreme peak — the next update arrives in seconds. In [Distributed Counters](/learn/system-design/questions/distributed-counters), approximate counts during overload are acceptable when exact precision isn't required. The [Rate Limiting](/learn/system-design/design-patterns/rate-limiting) pattern covers related strategies for controlling inbound request volume. ### Batching Every individual write carries overhead: a network round trip, a transaction begin/commit, index updates, and a WAL flush. If your app sends 1,000 tiny writes per second, you're paying that overhead 1,000 times. **Batching** groups multiple small writes into a single larger write. Instead of inserting rows one at a time, you collect them over a short window (say 100-500ms) and issue one `INSERT ... VALUES (...), (...), (...)` statement. The database does one transaction, one index update pass, and one WAL flush for the entire batch. Batching can happen at three different layers: **Application layer:** Your service batches up writes before sending to the database. This works well when the application isn't the source of truth — for example, a service reading from a Kafka topic can batch writes without risk of data loss since Kafka retains the events. **Intermediate processing:** A dedicated batcher service sits between the event stream and the database, aggregating events before forwarding them. Consider a system tracking "Like" counts. Instead of writing each like event individually, a batcher reads events over a window (say 1 minute), tabulates the changes per post, and forwards the aggregated counts to the database. If a post receives 100 likes in that window, you've reduced 100 writes to 1. **Database layer:** Most databases have configurable options for how often writes are flushed to disk. Redis's default configuration flushes writes to disk every 100ms. While elegant, this is the "big hammer" solution and should be reserved for extreme cases. One important caveat: batching only helps when multiple writes target the same data. If the majority of posts get 1 like per hour, a batch frequency of 1 minute provides zero benefit. Make sure batching actually matches your traffic patterns. **See it in practice:** In [Uber](/learn/system-design/questions/uber), driver location updates are batched over short intervals before writing to the geospatial store. In the [Ad Click Aggregator](/learn/system-design/questions/adds-click-aggregator), raw click events are micro-batched before persisting. The [Stream Processing & Realtime Aggregation](/learn/system-design/design-patterns/stream-processing-realtime-aggregation) pattern covers windowed aggregation in detail. ### Hierarchical Aggregation For the most extreme cases — live video streams, real-time analytics, massive fan-out systems — you often don't need to store individual events at all. You need aggregated views, and those views can be built up incrementally. **Hierarchical aggregation** processes data in stages, reducing volume at each step. Consider live video comments: millions of viewers can both comment and like comments on a stream. Whenever a viewer performs an action, all other viewers need to see it. This creates an ugly N-to-N problem — millions writing, millions reading. The solution is to add processing stages. Viewers are assigned to **write processors** using consistent hashing. Each write processor aggregates likes by comment ID over a short window and forwards batched updates to a **root processor**. The root processor merges updates and pushes them to **broadcast nodes**, each responsible for a subset of viewers. By aggregating with the write processors and distributing with the broadcast nodes, you've reduced the number of writes any single component handles — at the cost of some latency from the additional processing stages. **See it in practice:** In [Top K](/learn/system-design/questions/topk), hierarchical counting across distributed nodes reduces the write volume for tracking the most popular items. In [Distributed Counters](/learn/system-design/questions/distributed-counters), multi-level aggregation handles the fan-in problem for high-frequency counter updates. The [Stream Processing & Realtime Aggregation](/learn/system-design/design-patterns/stream-processing-realtime-aggregation) pattern covers multi-stage processing pipelines in depth. ## Common Problems and How to Solve Them ### Hot Shard Problem Even with a good partition key, some entities generate wildly disproportionate traffic. A viral post might receive millions of likes while most posts get dozens. All those likes hash to the same shard, creating a **hot shard** that's overwhelmed while other shards sit idle. The solution: **random suffix sharding**. Instead of writing to key Ad42, you write to Ad42:rand(0..N), distributing the writes across N different shards. At read time, you query all N suffixed keys and aggregate the results (SUM, COUNT, etc.). You can apply this in two ways: **Split all keys a fixed K times.** Every key gets K sub-keys (post1-0, post1-1, ..., post1-K). This is simple but increases your dataset size by K and multiplies read volume by K. If a small K brings workload back in line, you've solved the problem. **Split hot keys dynamically.** Only break keys into sub-keys when they're detected as hot. Writers keep local statistics and conditionally split writes for hot keys. Readers always check for sub-keys. This is more efficient but adds complexity around coordinating which keys are split. Both approaches work for data that can be aggregated (likes, views, counts) but don't work for data that must remain atomic (user profiles). Fortunately, the latter category rarely faces the same write pressure. ### Cross-Shard Queries Once data is sharded, any query that spans multiple shards becomes a **scatter-gather** operation: you send the query to every shard, collect the results, and merge them. This is slow and resource-intensive. Three strategies to minimize cross-shard queries: - **Co-locate related data:** Choose your partition key so that data accessed together lives on the same shard. User ID works well because most queries are scoped to a single user - **Denormalize:** Store copies of frequently-joined data on each shard. This duplicates data but eliminates cross-shard joins - **Maintain a global index:** For queries that must span shards (e.g., "find all orders with status=pending"), maintain a separate index service that maps query criteria to shard locations **See it in practice:** In [Google Calendar](/learn/system-design/questions/google-calendar), events are sharded by user ID so all of a user's events live on one shard — but cross-user meeting lookups require scatter-gather. In [Slack](/learn/system-design/questions/slack), messages are sharded by channel so channel queries are efficient, but cross-channel search requires a global index. The [Search & Full-Text Indexing](/learn/system-design/design-patterns/search-full-text-indexing) pattern covers how to build global indexes that span sharded data. ### Resharding As traffic patterns change, some shards grow faster than others. **Resharding** means moving data between shards to keep the load even. This is operationally painful. The naive approach is to take the system offline, rehash all data, and move it to new shards. But this creates hours of downtime for large datasets. Production systems use **gradual migration** with a dual-write phase: writes go to both the old shard and the new shard during migration. Reads prefer the new shard. This allows data to move without downtime, though it adds complexity. ## How to Use Write scaling patterns show up in many high-scale system design interviews. Don't wait for the interviewer to ask about it — proactively identify bottlenecks, validate them with back-of-the-envelope math, and propose solutions as deep dives. ### Common Interview Scenarios ### When NOT to Use Be careful of employing write scaling strategies when no scaling is necessary. If you see something that looks like a bottleneck, do some quick math to confirm. Each strategy comes with trade-offs: queues mean eventual consistency and delay, partitioning may compromise the read path, batching adds latency and moving pieces. Show your interviewer that you're cognizant of these trade-offs before making a proposal. The worst case is creating a problem where one doesn't exist. ### Key Takeaway Write scaling is about reducing throughput per component. Whether you're spreading 10,000 writes across 10 shards, smoothing bursts through queues, or batching them into 100 bulk operations, you're applying the same principle: make each individual component handle manageable load.