Module 3: Storage, Databases & Caching

Two Models, Two Philosophies

Every application stores data. The question is how that data is organized, queried, and scaled. For decades, relational databases (SQL) dominated. Then, around 2009, a wave of distributed systems gave rise to NoSQL. These are not competing religions. They are engineering tradeoffs. The right choice depends on your data shape, your access patterns, and your consistency requirements.

Relational Databases (SQL)

Relational databases were designed for correctness. A banking system transferring money between accounts needs a guarantee: debit one account and credit the other, or do neither. ACID transactions provide that guarantee. The schema enforces structure. You define columns and types upfront. If you try to insert a string into an integer column, the database rejects it. This strictness prevents entire categories of bugs.

SQL databases excel at complex queries. Joins allow you to combine data from multiple tables in a single query. An e-commerce system can answer "show me all orders from customers in Jakarta who bought more than 3 items last month" with one SQL statement. Try doing that efficiently in a key-value store.

The cost of this power is rigidity. Schema changes (adding a column, changing a type) require migrations. On large tables with billions of rows, migrations can take hours and lock the table. Horizontal scaling is difficult because joins across machines are expensive. Most relational databases scale vertically first, then add read replicas for read-heavy workloads.

Major relational databases: PostgreSQL, MySQL, Oracle, SQL Server, SQLite.

NoSQL Databases

NoSQL emerged to solve problems that relational databases handle poorly: massive write throughput across distributed clusters, highly variable data shapes, and workloads where horizontal scaling matters more than multi-table joins.

A document database like MongoDB stores JSON-like documents. Each document can have different fields. You do not need to run ALTER TABLE to add a new attribute. You just start writing documents with the new field. This flexibility accelerates early development and works well for data that does not fit neatly into rows and columns.

Most NoSQL databases achieve horizontal scalability by partitioning data across nodes. Cassandra, for example, distributes data using consistent hashing. Adding a node automatically rebalances part of the data. This makes scaling to hundreds of nodes straightforward, something relational databases struggle with.

The tradeoff is consistency. Most NoSQL systems offer eventual consistency by default: after a write, replicas will converge to the same state, but a read immediately after a write might return stale data. Some NoSQL databases (like MongoDB with majority write concern, or DynamoDB with strong reads) can provide stronger guarantees at the cost of latency.

Throughput Comparison

Benchmark numbers depend heavily on hardware, configuration, data size, and workload shape. The following chart uses approximate figures from YCSB (Yahoo Cloud Serving Benchmark) studies and vendor benchmarks to illustrate relative throughput ranges. These are order-of-magnitude comparisons, not absolute truths for your specific workload.

Redis dominates because it is an in-memory store. Cassandra's write throughput is high because its LSM-tree storage engine is optimized for sequential writes. PostgreSQL's numbers are lower on raw throughput but it handles complex queries that the others cannot. These numbers shift dramatically with indexing, replication, and query complexity.

Comparison Across Dimensions

Polyglot Persistence

Most real-world systems do not pick one database for everything. An e-commerce platform might use PostgreSQL for orders and inventory (transactions matter), Redis for session storage and caching (speed matters), Elasticsearch for product search (full-text indexing matters), and a graph database for recommendations (relationship traversal matters).

This is not over-engineering. It is recognizing that no single database excels at every workload. The cost is operational complexity: more systems to monitor, more failure modes, more data synchronization concerns.

The diagram above shows a polyglot architecture. PostgreSQL serves as the source of truth for transactional data. Change data capture (CDC) streams changes to Elasticsearch for search indexing and to Redis for cache warming. Each database handles the workload it was designed for.

The ACID vs. BASE Spectrum

SQL databases follow ACID. Most NoSQL databases follow BASE: Basically Available, Soft state, Eventually consistent. These are not binary categories. Many databases sit on a spectrum.

MongoDB now supports multi-document ACID transactions (since version 4.0). DynamoDB offers strongly consistent reads. CockroachDB is a distributed SQL database that provides ACID guarantees across multiple nodes. The boundaries between SQL and NoSQL are blurring. What matters is understanding the default behavior and what guarantees you are paying for in terms of latency and throughput.

Systems Thinking Lens

Choosing a database is a leverage point. Get it right early and the system scales naturally. Get it wrong and you spend months migrating under pressure. The feedback loop is delayed: you pick a database in month one, but the pain of a bad choice shows up in month twelve when traffic grows 10x.

Polyglot persistence creates a balancing loop. Each additional database solves one problem but introduces operational overhead. At some point, the cost of managing five different databases outweighs the performance benefit. The systems thinker asks: where is the point of diminishing returns for our team size and operational maturity?

Further Reading

• Martin Kleppmann, Designing Data-Intensive Applications (O'Reilly, 2017), Chapters 2 and 3. The definitive treatment of data models, storage engines, and the SQL vs. NoSQL landscape.
• Wikipedia, NoSQL. Comprehensive overview of the NoSQL movement, history, and taxonomy.
• Rick Houlihan, Amazon DynamoDB Deep Dive (AWS re:Invent 2018). Masterclass on NoSQL data modeling and access pattern design from the DynamoDB team.
• Percona, PostgreSQL Replication Guide. Practical guide to scaling PostgreSQL with replicas.
• benchANT, Database Performance Ranking. Independent benchmarks comparing database throughput and cost across cloud providers.

Four Families, Four Philosophies

Saying "NoSQL" tells you almost nothing about how a database actually works. It is like saying "not a car" when someone asks what vehicle you drive. It could be a motorcycle, a bus, a helicopter, or a bicycle. NoSQL is an umbrella term covering at least four fundamentally different data models, each optimized for different access patterns.

The four families are: key-value stores, document databases, wide-column stores, and graph databases. Picking the right family means understanding your data shape and how you will query it.

Key-Value Stores

Key-value stores are hash tables at scale. You give the database a key, it returns a value. You cannot query by value contents. You cannot filter or sort. You can only get, set, and delete by key. This constraint is the source of their power. With no secondary indexes, no joins, and no query parsing, operations complete in microseconds.

Redis is the most widely used key-value store. It holds data in memory, supports data structures beyond simple strings (lists, sets, sorted sets, hashes), and provides sub-millisecond latency. DynamoDB is Amazon's managed key-value and document store, built for horizontal scalability with single-digit millisecond reads at any scale.

Use when: Session storage, caching, rate limiting, leaderboards, feature flags. Any workload where the access pattern is "I know the key, give me the value."

Avoid when: You need to search by attributes within the value, or you need relationships between records.

Document Databases

Document databases sit between key-value simplicity and relational richness. Each document is a self-contained unit. A user document might contain the user's name, address (nested object), order history (array of objects), and preferences (nested map). You can query any of these fields.

MongoDB is the dominant document database. It supports rich queries, aggregation pipelines, secondary indexes, and multi-document ACID transactions. Couchbase and Amazon DocumentDB are other options in this space.

The document model works well when your data naturally forms aggregates. An e-commerce product is a good example: the product name, description, images, variants, and reviews form a logical unit that gets read and written together. Storing this as a single document avoids the five-table join you would need in a relational database.

Use when: Content management, product catalogs, user profiles with variable fields, event logging.

Avoid when: You need complex joins across document types or strict referential integrity between collections.

Wide-Column Stores

Wide-column stores were inspired by Google's Bigtable paper (2006). The data model looks like a table, but it is fundamentally different from relational tables. There is no fixed schema. Each row can have millions of columns. Columns are grouped into families, and the database stores each family contiguously on disk.

Apache Cassandra is the most prominent wide-column store. It provides linear write scalability: adding a node to the cluster increases write throughput proportionally. Cassandra partitions data by a hash of the partition key and sorts data within each partition by the clustering columns. This makes time-ordered queries within a partition very fast.

HBase (built on Hadoop) and ScyllaDB (a Cassandra-compatible database written in C++) are other notable wide-column stores.

Use when: Time-series data (IoT sensor readings, event logs), messaging at scale, write-heavy workloads with known query patterns.

Avoid when: You need ad-hoc queries, complex aggregations, or do not know your access patterns at design time.

Graph Databases

Relational databases can model graphs using join tables, but traversing deep relationships becomes exponentially expensive. A three-hop query (friends of friends of friends) requires three self-joins, each of which scans the entire relationship table. Graph databases use index-free adjacency: each node directly references its neighbors, so traversal cost is proportional to the local neighborhood size, not the total graph size.

Neo4j is the leading graph database. It uses the Cypher query language, which reads like a visual pattern: MATCH (a)-[:FOLLOWS]->(b)-[:FOLLOWS]->(c) RETURN c. Amazon Neptune and TigerGraph are other graph databases used at scale.

Use when: Social networks, recommendation engines, fraud detection, knowledge graphs, dependency mapping.

Avoid when: Your data has few relationships, or your queries are primarily simple lookups and range scans.

Data Model Comparison

Comparison Table

Radar Comparison

The radar chart below compares the four NoSQL families across five dimensions, scored on a relative scale of 1 (weakest) to 5 (strongest). These are directional, not absolute. Your mileage will vary depending on the specific product, configuration, and workload.

Key-value stores score the highest on speed and scalability but the lowest on query flexibility. Graph databases are the inverse: rich queries but harder to scale horizontally. Document and wide-column databases sit in different middle grounds. This is why polyglot persistence exists. No single family wins on every axis.

Choosing the Right Family

The decision tree above is a starting point, not a prescription. Many workloads could fit multiple families. When in doubt, start with the simplest option that meets your access pattern requirements. You can always add specialized databases later through polyglot persistence.

Systems Thinking Lens

Each NoSQL family creates a different constraint on your system. Key-value stores constrain your query patterns but free you from schema management. Graph databases free your query patterns but constrain your scaling options. These are balancing loops. The freedom you gain on one axis is paid for on another.

The leverage point is access pattern analysis. Before choosing a database family, write down your ten most important queries. If you cannot do that, you are not ready to choose. The database family should emerge from the access patterns, not the other way around. Too many teams pick a database because it is trendy, then spend months fighting its constraints.

Further Reading

• Fay Chang et al., Bigtable: A Distributed Storage System for Structured Data (Google, 2006). The paper that inspired wide-column stores including HBase and Cassandra.
• MongoDB Documentation, Data Modeling Introduction. Official guide to document modeling patterns in MongoDB.
• Neo4j Documentation, What is a Graph Database?. Clear introduction to graph data modeling and the property graph model.
• ScyllaDB, NoSQL Database Comparison. Detailed comparison across NoSQL families with performance characteristics.
• Redis Documentation, Data Types. Goes beyond simple key-value to show Redis's rich data structure support.

Why Shard?

A single database server has limits. CPU, memory, disk I/O, network bandwidth. Vertical scaling buys you time, but eventually you hit the ceiling. When your tables hold billions of rows and your write throughput exceeds what one machine can handle, you need to split the data across multiple machines. That is sharding.

Sharding is not the same as replication. Replication copies the same data to multiple machines for redundancy and read scaling. Sharding distributes different data to different machines for write scaling and storage capacity. Most production systems use both.

Three Sharding Strategies

Range-Based Sharding

Divide the shard key space into contiguous ranges. Users 1 through 1,000,000 go to Shard A. Users 1,000,001 through 2,000,000 go to Shard B. And so on.

Range-based sharding is simple to understand and implement. Range queries are efficient because all data within a range lives on the same shard. If you need all users with IDs between 500,000 and 600,000, that query hits a single shard.

The problem is hotspots. If user IDs are auto-incrementing, all new writes go to the highest shard. That shard becomes a bottleneck while the others sit idle. Time-based keys have the same problem: all recent data lands on the same shard.

Hash-Based Sharding

Apply a hash function to the shard key and use the result to determine the shard. For example: shard = hash(user_id) % number_of_shards. A good hash function distributes keys uniformly, so writes spread evenly across all shards.

Hash-based sharding eliminates hotspots for sequential keys. User IDs 1, 2, 3, 4 might map to shards 3, 1, 0, 2 respectively. The tradeoff: range queries become expensive. "Find all users with IDs 1 through 1000" now requires querying every shard and merging results, because those IDs are scattered.

Resharding is also painful. If you change the number of shards (say from 4 to 6), the modulo changes, and most keys map to different shards. You have to move data. Consistent hashing (covered in Session 3.6) mitigates this problem.

Directory-Based Sharding

Maintain a lookup table that maps each shard key (or key range) to a specific shard. The application checks the directory before every database operation to determine which shard to query.

Directory-based sharding offers maximum flexibility. You can place data on shards based on any criteria: geography, tenant ID, or custom business logic. You can move data between shards by updating the directory without changing the hash function or recalculating ranges.

The cost is the directory itself. It becomes a single point of failure and a latency bottleneck. Every query requires an extra lookup. The directory must be highly available, fast, and consistent. Many teams implement the directory in a cache (Redis) backed by a durable store.

Hash-Based Sharding in Action

In this diagram, the shard router computes hash(user_id) % 4 and routes the query to the appropriate shard. Each shard is a separate PostgreSQL instance holding roughly one quarter of all users. Writes are distributed evenly. Reads by user_id hit exactly one shard.

Shard Key Selection

A good shard key has three properties:

• High cardinality. The key should have many distinct values. Sharding by country gives you ~200 shards at most, and some shards (India, USA) will be much larger than others.
• Even distribution. Values should distribute data uniformly across shards. Auto-incrementing IDs with hash-based sharding achieve this. Sharding by last name does not (too many "Smith" entries on one shard).
• Query alignment. Your most frequent query should include the shard key. If 90% of queries are "get all orders for user X," then user_id is a good shard key because each query hits one shard.

Strategy Comparison

Cross-Shard Queries

The hardest problem in sharding is queries that span multiple shards. If your data is sharded by user_id but you need to "find all users in Jakarta," the city field is not the shard key. The query must hit every shard, collect partial results, and merge them. This is called a scatter-gather query.

Scatter-gather queries are slow and resource-intensive. They scale linearly with the number of shards. With 4 shards, it is tolerable. With 400 shards, it is a serious performance problem.

Solutions include maintaining secondary indexes (either global or per-shard), denormalizing data into a separate lookup table, or using a search engine like Elasticsearch for cross-shard queries. Each solution adds complexity and consistency challenges.

Real-World Examples

Instagram: Sharding PostgreSQL

Instagram needed to handle over 25 photo uploads and 90 likes per second in its early days. They chose to shard PostgreSQL rather than switch to NoSQL. Their approach used thousands of "logical" shards mapped to a smaller number of physical PostgreSQL instances. Each logical shard was a PostgreSQL schema containing the sharded tables. This allowed them to move logical shards between physical servers without re-bucketing data.

For globally unique IDs, Instagram designed a custom ID generation scheme using PostgreSQL's PL/pgSQL. Each ID encodes 41 bits for timestamp, 13 bits for logical shard ID, and 10 bits for an auto-incrementing sequence. This produces IDs that are sortable by time and unique across all shards.

Discord: Cassandra to ScyllaDB

Discord stored trillions of messages in Cassandra, partitioned by channel_id. This worked well for most channels, but large public channels (with millions of messages) created hot partitions. Cassandra's garbage collector caused latency spikes under this workload. Discord eventually migrated to ScyllaDB (a Cassandra-compatible database written in C++) and added a data services layer in Rust with request coalescing to handle hot partitions. The migration reduced their cluster from 177 Cassandra nodes to 72 ScyllaDB nodes, with p99 read latency dropping from 40-125ms to 15ms.

Instagram's logical shard abstraction is the key insight. By separating logical partitioning from physical placement, resharding becomes a data migration problem rather than a re-hashing problem. You move schemas between servers without touching the application's routing logic.

Systems Thinking Lens

Sharding introduces a reinforcing complexity loop. More data requires more shards. More shards increase the surface area for cross-shard queries and operational issues. Operational issues require more engineering effort. More engineering effort means less time for feature development. This loop accelerates as the system grows.

The leverage point is the shard key. A well-chosen shard key keeps most queries within a single shard, which limits the impact of the complexity loop. Instagram's choice to shard by user_id aligned with their primary access pattern (show me this user's photos), keeping the complexity manageable even at massive scale.

Further Reading

• Instagram Engineering, Sharding & IDs at Instagram. How Instagram designed their sharding scheme and ID generation for PostgreSQL.
• Discord Engineering, How Discord Stores Trillions of Messages. The journey from MongoDB to Cassandra to ScyllaDB, with details on hot partition problems.
• PlanetScale, Sharding Strategies: Directory-Based, Range-Based, and Hash-Based. Clear comparison of the three sharding strategies with practical guidance.
• Martin Kleppmann, Designing Data-Intensive Applications (O'Reilly, 2017), Chapter 6 ("Partitioning"). Thorough treatment of partitioning strategies and rebalancing.
• Instagram Engineering, Handling Growth with Postgres: 5 Tips from Instagram. Practical tips for scaling PostgreSQL at Instagram's growth rate.

Why Replicate?

A single database server is a single point of failure. If it goes down, your application goes down. Replication copies data from one server to others, providing two benefits: fault tolerance (if one server dies, another has the data) and read scaling (distribute read queries across multiple copies).

Replication and sharding solve different problems. Sharding splits data to handle more writes and store more data. Replication copies data to handle more reads and survive failures. Most production databases use both: shard the data across multiple primaries, then replicate each shard to one or more replicas.

Primary-Replica Replication

The most common replication topology. All writes go to a single primary server. The primary streams changes (via a write-ahead log, binary log, or oplog) to one or more replica servers. Replicas apply these changes and serve read queries.

This topology is straightforward. There is no write conflict because only one server accepts writes. Replicas are read-only, which means you can add replicas to scale read throughput linearly. Three replicas handle roughly three times the read queries of a single server.

The downside is write scalability. All writes funnel through one server. If your write throughput exceeds what one machine can handle, primary-replica replication alone will not help. You need sharding for write scaling.

Failover is the other concern. If the primary dies, one of the replicas must be promoted to primary. This can be automated (most managed databases do this) but involves a brief window where writes are unavailable. The promoted replica may also be slightly behind the old primary, meaning some recent writes could be lost in asynchronous replication setups.

Multi-Primary Replication

Multi-primary replication is used when you need to accept writes in multiple geographic regions. A user in Jakarta writes to the Jakarta primary. A user in Singapore writes to the Singapore primary. Each primary sends its changes to the other. Reads are served locally, and writes do not need to cross the ocean before being acknowledged.

The fundamental problem is conflicts. If user A updates their email on the Jakarta primary and user B updates the same email field on the Singapore primary at the same time, which write wins? Conflict resolution strategies include:

• Last-write-wins (LWW): The write with the later timestamp wins. Simple but loses data silently. Clock synchronization across data centers is imperfect.
• Application-level resolution: The database stores both versions and lets the application decide. CouchDB and DynamoDB use this approach.
• CRDTs (Conflict-free Replicated Data Types): Data structures designed to merge automatically without conflicts. Works for counters, sets, and some text editing scenarios but not for arbitrary data.

Synchronous vs. Asynchronous Replication

When the primary commits a write, it can wait for replicas to confirm before acknowledging the client (synchronous), or it can acknowledge immediately and let replicas catch up later (asynchronous). This is a fundamental tradeoff between durability and latency.

Most production systems use asynchronous replication because the latency penalty of synchronous replication is too high, especially across data centers. A write that takes 2ms on the primary might take 50-200ms if it must wait for a replica in another region. Semi-synchronous replication is a common compromise: wait for at least one replica to confirm, then acknowledge the client.

Replication Lag

Replication lag causes stale reads. A user updates their profile on the primary, then immediately loads the profile page. If the read is routed to a replica that has not yet received the update, the user sees old data. This is the "read-your-own-writes" problem.

Lag is not constant. During normal operation, it might be 10-50 milliseconds. During a traffic spike that doubles write throughput, the replica's single-threaded apply process falls behind. Long-running analytical queries on replicas can also cause lag by competing for I/O and CPU resources.

The chart above simulates typical replication lag over a 24-hour period. During off-peak hours (midnight to 6 AM), lag stays under 20ms. During peak load (late morning and evening), lag spikes above the 200ms threshold. These spikes correlate with write volume: when the primary processes more writes per second, the replica falls further behind.

Handling Replication Lag

Several strategies mitigate the effects of replication lag:

• Read-your-own-writes: After a user writes data, route that user's subsequent reads to the primary (or a replica known to be up-to-date) for a short window. This ensures users see their own changes immediately.
• Monotonic reads: Ensure a user always reads from the same replica. This prevents the confusing experience of seeing newer data, then older data on the next request (because a different, more-lagged replica served it).
• Lag-aware routing: Monitor replication lag on each replica. Route reads only to replicas within an acceptable lag threshold. If all replicas are too far behind, fall back to the primary.
• Causal consistency: Track causal dependencies between writes and reads. A read that depends on a specific write waits until the replica has applied that write before returning results.

Replication Strategy Comparison

Systems Thinking Lens

Replication creates a fundamental balancing loop between consistency and performance. Adding replicas increases read throughput (reinforcing loop), but each replica introduces replication lag (balancing loop). The faster you write, the further behind replicas fall, and the more stale reads your users experience.

The leverage point is not the replication strategy itself but the read/write ratio of your workload. If your application is 95% reads and 5% writes, primary-replica replication with asynchronous replication works well. The lag affects only a small fraction of operations. If your application is 50% writes, replication lag becomes a dominant concern, and you may need synchronous replication or architectural changes (like event sourcing) that decouple reads from writes entirely.

Multi-primary replication has a hidden reinforcing loop: the more primaries you add, the more conflict resolution logic you need, the more edge cases emerge, the more engineering time you spend debugging subtle data inconsistencies instead of building features.

Further Reading

• Martin Kleppmann, Designing Data-Intensive Applications (O'Reilly, 2017), Chapter 5 ("Replication"). The most thorough treatment of replication topologies, consistency models, and lag handling.
• Percona, Replication Lag in PostgreSQL. Practical guide to measuring, monitoring, and reducing PostgreSQL replication lag.
• CockroachDB, Synchronous and Asynchronous Replication Explained. Clear explanation of how data loss occurs during outages with async replication.
• AWS, Troubleshoot High Replica Lag with Amazon RDS for MySQL. Real-world troubleshooting steps for replication lag in managed databases.
• Percona, PostgreSQL Replication with Asynchronous and Synchronous Standbys. Hands-on configuration guide for both replication modes in PostgreSQL.

Why Indexes Exist

Without an index, every query is a full table scan. The database reads every row, checks if it matches your condition, and discards what does not. For a table with 100 rows, this is fine. For a table with 10 million rows, it is catastrophic.

An index is a separate data structure that maps column values to their physical locations in the table. Think of it like the index at the back of a textbook: instead of reading every page to find "B-tree," you look up "B-tree" in the index and jump directly to page 247.

B-Tree Indexes

The B-tree (specifically B+ tree) is the default index type in PostgreSQL, MySQL, SQL Server, and virtually every relational database. It organizes data in a balanced tree structure where every leaf node is the same distance from the root.

A B+ tree with a branching factor of 500 can index 8 TB of data with only 4 levels. That means any lookup requires at most 4 disk I/O operations, regardless of table size. The time complexity is O(log n), which grows very slowly.

B-trees support equality lookups (WHERE id = 42), range queries (WHERE price BETWEEN 10 AND 50), prefix matching (WHERE name LIKE 'Jo%'), and sorting (ORDER BY created_at). This versatility is why they are the default.

The root node contains separator keys that direct the search. Internal nodes narrow the range. Leaf nodes contain the actual pointers to table rows. In a B+ tree, leaf nodes are also linked together, which makes range scans efficient: once you find the starting point, you walk the linked list instead of re-traversing the tree.

Hash Indexes

A hash index applies a hash function to the column value and stores the result in a hash table. Lookups are O(1) on average, which is faster than B-tree's O(log n). PostgreSQL benchmarks show hash indexes can be 10-22% faster than B-trees for pure equality lookups.

The tradeoff is severe: hash indexes support only equality comparisons (WHERE id = 42). They cannot do range queries, sorting, or prefix matching. If you need any of those operations on the column, a hash index is useless.

GIN and GiST Indexes

PostgreSQL offers two specialized index types for complex data. GIN (Generalized Inverted Index) is built for full-text search, arrays, and JSONB. It creates an entry for every element inside a composite value. GIN lookups are about 3x faster than GiST for text search, but GIN indexes take about 3x longer to build.

GiST (Generalized Search Tree) handles geometric data, ranges, and nearest-neighbor queries. GiST indexes are lossy for text search (they may return false positives that require a recheck), but they update faster than GIN, making them better for frequently changing data.

Index Comparison

Composite Indexes and Column Order

A composite index includes multiple columns. The database sorts entries first by the leftmost column, then by the second column within each group, and so on. This ordering has a critical consequence called the leftmost prefix rule.

Consider an index on (user_id, status, created_at). This index efficiently supports:

• WHERE user_id = 5
• WHERE user_id = 5 AND status = 'pending'
• WHERE user_id = 5 AND status = 'pending' ORDER BY created_at

It does NOT efficiently support WHERE status = 'pending' alone, because status is not the leftmost column.

Index Selectivity

Selectivity measures how well an index narrows down results. A column with 1 million unique values in a 1 million row table has selectivity of 1.0 (perfect). A boolean column with only two values has selectivity close to 0 (terrible).

Put highly selective columns first in composite indexes. If user_id narrows 10 million rows to 50, and status narrows them to 5 million, always lead with user_id.

Covering Indexes

A covering index contains all columns needed by a query. The database can answer the query entirely from the index without touching the table data. This eliminates the random I/O cost of jumping from the index to the table rows.

In PostgreSQL, you use INCLUDE to add non-searchable columns to the index:

CREATE INDEX idx_orders_user_status
ON orders (user_id, status)
INCLUDE (created_at, total);

Now a query selecting user_id, status, created_at, total with a WHERE on user_id and status never touches the heap. The savings are substantial for wide tables with many columns.

The Hidden Write Cost

Every index must be updated on every write operation. An INSERT into a table with 5 indexes means 6 write operations: one for the table and one for each index. An UPDATE that modifies an indexed column triggers an index delete and an index insert for each affected index.

This cost is not theoretical. On write-heavy workloads, excessive indexing is a common cause of degraded performance. The solution is to index deliberately: only create indexes that serve actual queries, and drop indexes that query plans never use.

Query Time vs. Table Size

The following chart shows how query response time changes as table size grows, comparing a full table scan against an indexed lookup.

The table scan time grows linearly with row count. The indexed lookup barely moves. At 10 million rows, the difference is four orders of magnitude.

Systems Thinking Lens

Indexes are a classic balancing loop. Adding an index improves read performance, which encourages more queries, which eventually leads to more indexes, which degrades write performance. The leverage point is not "add more indexes" but "understand your query patterns." A single well-designed composite index often replaces three or four single-column indexes while also reducing write overhead.

Further Reading

• MySQL Reference Manual, Comparison of B-Tree and Hash Indexes. Official MySQL documentation on when to use each index type.
• PlanetScale, Composite Indexes. Practical guide to composite index design with the leftmost prefix rule.
• pganalyze, Understanding Postgres GIN Indexes. Deep dive into GIN index internals, build time, and query performance.
• PostgreSQL Documentation, Preferred Index Types for Text Search. Official comparison of GIN vs GiST for full-text search.
• EnterpriseDB, Are Hash Indexes Faster than B-Tree Indexes in Postgres?. Benchmark results comparing hash and B-tree index performance.

SELECT * FROM orders
WHERE user_id = ?
  AND status = 'pending'
ORDER BY created_at DESC;

The Problem with Modulo Hashing

Distributed systems often need to assign data to nodes. The simplest approach is modulo hashing: take the hash of a key, divide by the number of nodes, and use the remainder as the node assignment. node = hash(key) % N.

This works well as long as N never changes. The moment you add or remove a node, nearly every key maps to a different node. If you go from 4 nodes to 5, the formula changes from hash(key) % 4 to hash(key) % 5. For most keys, the remainder is different. In a cache cluster, this means almost every cached item is now on the wrong node. The cache hit rate drops to nearly zero. Every request hits the database. You just turned a scaling event into a cascading failure.

Consistent Hashing

Consistent hashing solves this by mapping both nodes and keys onto the same circular hash space (a "hash ring"). Each node is placed at a position on the ring determined by hashing its identifier. Each key is placed on the ring by hashing its value. The key is assigned to the first node encountered when walking clockwise from the key's position.

When a node is added, it takes over a segment of the ring from its clockwise neighbor. Only the keys in that segment move. When a node is removed, its segment is absorbed by the next clockwise node. Only the keys assigned to the removed node need to move.

In this ring, Key 1 at position 45 degrees walks clockwise and hits Node B at 90 degrees. Key 4 at position 350 degrees wraps around and hits Node A at 0 degrees. Each key belongs to the next node clockwise.

Key Redistribution: The Numbers

The difference between modulo hashing and consistent hashing becomes dramatic when nodes change.

With modulo hashing, the problem gets worse as you scale. Going from 99 to 100 nodes moves 99% of keys. With consistent hashing, the same change moves only 1%. The gap widens at every scale.

Comparison Table

The Problem with Basic Consistent Hashing

With only one position per node on the ring, load distribution is uneven. If Node A sits at 0 degrees and Node B sits at 10 degrees, Node B owns a tiny 10-degree arc while Node A owns a massive 350-degree arc. Node A gets roughly 35x more traffic than Node B. Random placement does not guarantee even spacing.

Removing a node makes this worse. The entire load of the removed node shifts to a single neighbor, potentially doubling that neighbor's load instantly.

Virtual Nodes

Virtual nodes (vnodes) solve load imbalance by giving each physical node multiple positions on the ring. Instead of hashing a node once, you hash it with different suffixes: hash("NodeA-1"), hash("NodeA-2"), hash("NodeA-3"), and so on. Each virtual node is a separate point on the ring, but they all map back to the same physical server.

With 3 virtual nodes per physical node, each physical node owns 3 arcs on the ring instead of 1. The arcs are spread across the ring, so the load is more evenly distributed. When Node B is removed, its three arcs are absorbed by three different nodes, not just one. This prevents any single node from being overwhelmed.

Virtual Nodes Enable Heterogeneous Clusters

Not all servers are equal. A machine with 64 GB of RAM can handle more data than one with 16 GB. Virtual nodes let you assign more vnodes to more powerful machines. If Server A has 4x the capacity of Server B, give Server A 200 virtual nodes and Server B 50. Server A will own roughly 4x as much of the ring.

This is how DynamoDB and Cassandra handle mixed hardware in production clusters. The vnode count per node is a configuration parameter that can be adjusted without changing the hashing algorithm.

Systems Thinking Lens

Modulo hashing is a brittle system with tight coupling between node count and key assignment. Any change to one variable (node count) causes a cascade through the entire system (mass key migration). Consistent hashing decouples these variables. The hash ring acts as an abstraction layer that absorbs changes locally instead of propagating them globally.

Virtual nodes add a second layer of decoupling: physical capacity from ring position. This makes the system adaptive. You can scale heterogeneously, handle failures gracefully, and rebalance without coordination storms.

Further Reading

• David Karger et al., Consistent Hashing and Random Trees (STOC 1997). The original paper that introduced consistent hashing.
• AlgoMaster, Consistent Hashing Explained. Clear walkthrough of hash rings, virtual nodes, and redistribution math.
• GeeksforGeeks, Consistent Hashing in System Design. Visual explanation with diagrams and real-world applications.
• Hello Interview, Consistent Hashing for System Design. Interview-focused treatment with practice scenarios.

Three Kinds of Storage

When engineers say "storage," they might mean three fundamentally different things. Block storage gives you raw disk volumes. File storage gives you a hierarchical file system. Object storage gives you a flat namespace of immutable blobs with metadata. Each model serves different workloads, and choosing wrong means paying too much, building unnecessary complexity, or hitting walls that should not exist.

Object Storage: S3-Style

Amazon S3 is the canonical object store, and its API has become a de facto standard (MinIO, Cloudflare R2, and Backblaze B2 all implement S3-compatible APIs). The model is simple: you have buckets (containers) and objects (files with metadata). Every object has a unique key within its bucket. You interact with it over HTTP.

Objects in S3 are replicated across at least 3 availability zones automatically. The durability guarantee of 99.999999999% means that if you store 10 million objects, you can statistically expect to lose one every 10,000 years. No database offers this durability at this cost.

Storage Cost Comparison

The price difference between storage tiers is dramatic. Choosing the right tier based on access frequency saves substantial money at scale.

EBS block storage costs 80x more per GB than Glacier Deep Archive. Even S3 Standard is 3.5x cheaper than EBS. The tradeoff is access latency: EBS gives you sub-millisecond random reads, while Glacier Deep Archive can take 12 hours to retrieve data. Choose your tier based on how often you access the data, not how important it is.

Pre-Signed URLs

A pre-signed URL is a time-limited URL that grants temporary access to a private S3 object. Your backend generates the URL using its AWS credentials, and the client uses it to upload or download directly from S3. The backend never touches the file bytes.

This pattern has three advantages. First, the file upload goes directly from the client to S3, so your backend servers do not handle large file transfers. This saves bandwidth, memory, and CPU. Second, the pre-signed URL expires, so even if it leaks, the exposure window is short. Third, S3 handles multipart uploads, retries, and storage durability. Your backend stays simple.

For downloads, the same pattern applies. The backend generates a pre-signed GET URL, and the client downloads directly from S3. To serve files even faster, put a CDN (CloudFront, Cloudflare) in front of the S3 bucket. The CDN caches popular objects at edge locations close to users, reducing latency from hundreds of milliseconds to single digits.

CDN Integration

The standard architecture for serving static assets at scale combines object storage with a CDN:

On the first request, the CDN edge fetches the object from S3 and caches it. Subsequent requests for the same object are served from the edge, which is geographically close to the user. For popular content (profile photos, product images, CSS files), the cache hit rate typically exceeds 95%. You pay S3 retrieval costs only on cache misses.

Distributed File Systems: HDFS and GFS

For big data workloads (log processing, analytics, machine learning training), the access pattern is different. You need to read and write very large files (gigabytes to terabytes) sequentially, and you want computation to move to the data rather than the other way around.

HDFS was inspired by Google's GFS paper (2003). The design optimizes for throughput over latency: sequential reads of entire blocks are fast, but random access to individual bytes is not supported. The NameNode is a single point of failure (mitigated by standby NameNodes in production), and the system is designed for append-only writes. You do not edit files in place.

HDFS and object storage serve different needs. Object storage is for serving individual files to many users (a profile photo, a PDF). HDFS is for processing massive datasets across a compute cluster. In modern architectures, the line is blurring. Systems like Apache Iceberg and Delta Lake store data as Parquet files in S3, combining the durability of object storage with the query patterns of big data.

Systems Thinking Lens

Storage is a system with competing feedback loops. The cost loop pushes data toward colder, cheaper tiers. The latency loop pulls frequently accessed data toward hotter, faster tiers. The durability loop demands replication, which multiplies cost. A well-designed storage strategy does not pick one tier. It classifies data by access pattern and moves it through tiers automatically (S3 Intelligent-Tiering does this for $0.0025 per 1,000 objects per month).

The leverage point is lifecycle policy, not storage selection. The decision of where data starts is less important than the rules that govern where it moves over time.

Further Reading

• Amazon Web Services, Amazon S3 Pricing. Current pricing for all S3 storage classes, requests, and data transfer.
• Sanjay Ghemawat, Howard Gobioff, Shun-Tak Leung, The Google File System (SOSP 2003). The paper that inspired HDFS and modern distributed file systems.
• CloudForecast, Amazon S3 Pricing Guide. Detailed breakdown of S3 cost components including hidden costs like request fees and retrieval charges.
• AWS Documentation, Uploading Objects Using Pre-Signed URLs. Official guide to generating and using pre-signed URLs for direct client uploads.

Why Cache the Read Path

Most applications are read-heavy. A typical e-commerce product page is read 5,000 times for every update. A user profile is read hundreds of times between edits. If every read hits the database, you are doing redundant work. The data has not changed, but the database parses the query, searches the index, fetches the row, and serializes the result every single time.

A cache stores the result of a previous computation so it can be reused. For read-heavy workloads, caching reduces database load, cuts response time from milliseconds to microseconds, and lets you serve more traffic without scaling the database.

There are two primary patterns for caching on the read path: cache-aside and read-through. They differ in who is responsible for populating the cache.

Cache-Aside (Lazy Loading)

In cache-aside, the application manages the cache directly. The cache sits beside the database, not in front of it. The application checks the cache first. On a hit, it returns the cached data. On a miss, it queries the database, writes the result to the cache, and returns it.

Cache-aside has several strengths. The cache only contains data that has actually been requested, so you do not waste memory caching things nobody reads. The application has full control over cache keys, TTLs, and invalidation logic. And because the cache and database are independent, a cache failure does not bring down the application. Reads fall back to the database, which is slower but functional.

The weakness is the cold start problem. When the cache is empty (after a restart or deployment), every request is a cache miss. All traffic hits the database simultaneously. This is the thundering herd problem.

Read-Through

In read-through, the cache itself is responsible for loading data from the database. The application only talks to the cache. If the cache does not have the data, it fetches it from the database, stores it, and returns it. The application never queries the database directly for cached entities.

Read-through simplifies the application code. The application does not contain any cache-miss logic or database fallback code. The cache provider handles everything. This is particularly useful when multiple services need the same caching behavior. Instead of implementing cache-miss handling in each service, the cache provider handles it once.

Read-through also avoids one specific problem with cache-aside: duplicate database queries on concurrent misses. In cache-aside, if 100 requests arrive for the same uncached key simultaneously, all 100 may query the database before any of them writes to the cache. Read-through implementations typically use internal locking: the first request triggers a database fetch, and subsequent requests for the same key wait for the result instead of making redundant database calls.

Refresh-Ahead

Refresh-ahead is a proactive strategy that reloads cached data before it expires. The cache tracks access patterns and pre-fetches entries that are likely to be requested again. If a cached item with a 300-second TTL is accessed at the 250-second mark, the cache proactively refreshes it in the background so it never actually expires.

This eliminates the latency spike that occurs when a popular cache entry expires and the next request must wait for a database query. The downside is that it can waste resources refreshing data that nobody will request again.

Strategy Comparison

TTL Strategy

Time-to-live (TTL) determines how long a cached entry remains valid. Set it too short, and you get frequent cache misses that negate the benefit. Set it too long, and users see stale data.

The right TTL depends on two things: how often the underlying data changes and how much staleness your users can tolerate. A product catalog that updates twice a day can tolerate a 5-minute TTL. A stock price that changes every second needs a TTL under 1 second, or you should not cache it at all.

A practical approach: start with the ratio of reads to writes.

• 10,000:1 read/write ratio (product pages): TTL of 5-15 minutes. The data rarely changes, and even stale data is acceptable for a few minutes.
• 100:1 read/write ratio (user profiles): TTL of 1-5 minutes with explicit invalidation on write.
• 10:1 read/write ratio (inventory counts): TTL of 10-30 seconds, or skip caching and use database read replicas instead.

Cache Hit Ratio

The cache hit ratio measures what percentage of requests are served from the cache. Production systems typically target 90-99% hit ratios. Below 80%, the cache is not providing enough benefit to justify its operational cost. Above 95%, you are in excellent shape.

Hit ratio depends on the working set size relative to cache capacity, the access pattern (power-law distributions cache well, uniform distributions do not), and TTL settings. Monitor this metric continuously. A sudden drop in hit ratio usually indicates either a change in traffic patterns or a cache configuration problem.

Systems Thinking Lens

Caching introduces a balancing loop: as database load increases, you add caching, which reduces database load. But caching also introduces a new reinforcing loop: as the cache absorbs more traffic, the application becomes dependent on it. A cache failure that was tolerable at 50% hit ratio becomes catastrophic at 99% hit ratio, because the database has been sized for 1% of the traffic, not 100%.

The leverage point is not the cache itself but the invalidation strategy. A cache that serves stale data is worse than no cache, because it creates bugs that are intermittent and hard to reproduce. Design the invalidation path with the same rigor as the caching path.

Further Reading

• AWS Whitepapers, Database Caching Strategies Using Redis. Official AWS guide covering cache-aside, read-through, and write-through patterns with Redis.
• CodeAhoy, Caching Strategies and How to Choose the Right One. Practical comparison of all major caching strategies with decision criteria.
• NCache, Read-Through, Write-Through, Write-Behind Caching. Detailed explanation of cache provider-managed patterns with code examples.
• System Overflow, Cache-Aside Pattern. Deep dive into cache-aside implementation including thundering herd mitigation.

The Write Side of Caching

Session 3.8 covered the read path: cache-aside, read-through, and refresh-ahead. Those strategies answer one question: how does data get into the cache when someone reads it? This session covers the other half: what happens when someone writes new data?

The write path is where the hard trade-offs live. Every write creates two copies of truth: one in the cache, one in the database. The strategy you pick determines which copy updates first, whether they update together, and what happens when something fails between the two writes.

There are three core patterns: write-through, write-behind (also called write-back), and write-around. Each makes a different bet about what matters most.

Write-Through

In write-through caching, every write goes to both the cache and the database synchronously. The application writes to the cache, and the cache immediately writes to the database. The write is only acknowledged as successful after both stores have been updated.

The guarantee is strong: cache and database are always consistent. If the database write fails, the entire operation fails, and the cache is not updated. There is no window where the cache holds data that the database does not.

The cost is latency. Every write now has the latency of two operations: writing to cache (sub-millisecond) plus writing to the database (typically 1-10ms for a relational database). For read-heavy workloads where writes are infrequent, this penalty is acceptable. For write-heavy workloads, it adds up quickly.

Write-Behind (Write-Back)

Write-behind flips the priority. The application writes to the cache, and the cache acknowledges immediately. The database is updated asynchronously, sometime later. The delay might be a few milliseconds, a few seconds, or batched into periodic flushes.

Write latency drops dramatically because the application only waits for the cache write. The database sees fewer, larger batch writes instead of many small ones, which can significantly reduce I/O pressure. If your application writes 1,000 updates per second to the same key, write-behind can collapse those into a single database write.

The risk is data loss. If the cache node crashes before the async write reaches the database, those pending writes are gone. The database is now behind. For some systems this is fine. For others, it is catastrophic.

Write-Around

Write-around skips the cache entirely on writes. The application writes directly to the database. The cache is only populated when data is read (through a read strategy like cache-aside or read-through).

Write-around avoids cache pollution. If you write data that is unlikely to be read soon (log entries, audit records, analytics events), putting it in the cache wastes memory and may evict more frequently accessed data. By writing only to the database, the cache stays focused on hot data.

The downside is that a read immediately after a write will always be a cache miss. The data exists in the database but not in the cache, so the first read pays the full database round-trip cost. For workloads where writes are rarely followed by immediate reads, this is efficient. For workloads where "write then read" is common, write-around creates unnecessary latency.

Comparing the Three Strategies

Combining Strategies

Production systems rarely use a single write strategy for everything. A banking application might use write-through for account balances (consistency is non-negotiable), write-behind for session activity tracking (high volume, eventual consistency is fine), and write-around for transaction audit logs (written once, read rarely).

The decision tree is straightforward. Ask three questions about each data type:

1. Can you tolerate any data loss? If no, eliminate write-behind.
2. Do reads typically follow writes? If yes, eliminate write-around.
3. Is write throughput a bottleneck? If yes, consider write-behind with a durable queue (like Redis with AOF persistence) to reduce the data loss risk.

Systems Thinking Lens

Each write strategy creates a different feedback loop. Write-through creates a tight, synchronous loop: every write immediately confirms both stores are consistent. The system is predictable but rigid. Write-behind introduces delay into the loop. The application "thinks" the write succeeded, but the database has not confirmed it yet. This delay is a stock of pending writes that can grow if the database slows down. If that stock grows faster than it drains, the system eventually fails. Write-around decouples write and read paths entirely, creating two separate loops that only interact when a cache miss triggers a database read.

The key insight: consistency is not free. Every write strategy pays for it somewhere, either in latency, risk, or complexity. Your job is to decide where the payment is least painful for each data type in your system.

Further Reading

• Caching Strategies and How to Choose the Right One, CodeAhoy
• Caching Strategy: Write-Behind (Write-Back) Pattern, EnjoyAlgorithms
• Performance at Scale with Amazon ElastiCache, AWS Whitepapers
• Write-through vs Write-back vs Write-around: Trade-offs, Design Gurus

Two Tools, Different Philosophies

Redis and Memcached are both in-memory key-value stores. Both deliver sub-millisecond latency. Both are open-source and battle-tested at massive scale. But they were built with fundamentally different design goals, and those differences matter when you are choosing one for a real system.

Memcached was built to do one thing well: cache simple key-value pairs with maximum throughput. Redis was built as a data structure server that happens to be excellent at caching. That difference in ambition shapes everything from data model to persistence to operational complexity.

Redis: The Data Structure Server

Redis stores more than strings. Its native data types include strings, hashes, lists, sets, sorted sets, bitmaps, hyperloglogs, and streams. Each type comes with atomic operations. You can increment a counter, push to a list, add to a sorted set by score, or compute set intersections, all in a single command without client-side logic.

This matters because it reduces round trips. Instead of fetching a value, modifying it in your application, and writing it back (three operations with race conditions), you issue a single ZINCRBY or HINCRBY command. The operation is atomic. No locks needed.

Redis also offers persistence through two mechanisms. RDB (Redis Database) takes point-in-time snapshots at configurable intervals. AOF (Append Only File) logs every write operation and can replay them on restart. You can use both together: RDB for fast recovery, AOF for minimal data loss. With AOF set to fsync every second, you lose at most one second of writes on a crash.

Additional features include pub/sub messaging, Lua scripting for server-side logic, transactions with MULTI/EXEC, and Redis Streams for event log workloads similar to a lightweight Kafka.

Memcached: The Pure Cache

Memcached stores strings. That is it. Every value is a blob of bytes up to 1MB. There are no data structures, no persistence, no scripting, no pub/sub. When a Memcached server restarts, all data is gone.

This simplicity is a feature. Memcached is multi-threaded by default, using its slab allocator to handle concurrent reads and writes across all available CPU cores. Redis, historically single-threaded for command execution (though Redis 6+ introduced I/O threading for network operations), cannot match Memcached's throughput on a single node when the workload is many concurrent clients doing simple GET/SET operations on multiple cores.

Memcached's memory management is also predictable. Its slab allocator pre-divides memory into fixed-size chunks, which eliminates fragmentation. Redis uses jemalloc and can fragment under certain workloads (many small keys created and deleted), though in practice this is manageable with activedefrag.

Feature Comparison

Performance: Throughput and Latency

Benchmarks are always context-dependent, but industry testing gives us useful baselines. On a standard Linux server, Redis handles roughly 1.2 to 1.5 million operations per second for simple GET/SET workloads using pipelining. Memcached, with its multi-threaded architecture, achieves 1.0 to 1.2 million operations per second on similar hardware, with an advantage in highly concurrent multi-core scenarios.

For latency, both deliver sub-millisecond response times. Redis typically achieves around 0.1 to 0.15ms for simple GETs. Memcached is in the same range, around 0.15 to 0.25ms. At the P99 level under heavy concurrent load, Memcached's multi-threading can show lower tail latencies because it does not serialize all commands through a single thread.

The practical takeaway: for simple caching of string values with high concurrency, Memcached may edge ahead. For anything requiring data structures, persistence, or atomic operations on complex types, Redis is the only option.

Eviction Policies

When memory is full, the cache must decide what to remove. Memcached uses LRU (Least Recently Used) and that is your only choice. Redis provides eight eviction policies:

Cache Stampede: The Thundering Herd

A cache stampede happens when a popular cache key expires and many concurrent requests simultaneously discover the cache miss. All of them hit the database at once, potentially overwhelming it. If the database slows down or crashes, the problem cascades: more requests time out, more retries stack up, and the system spirals.

Three proven prevention techniques:

1. Distributed locking. When a cache miss occurs, the first request acquires a lock (using Redis SET key NX EX) and fetches from the database. All other requests wait for the lock to release, then read from the refreshed cache. Only one database query executes.

2. Probabilistic early expiration (XFetch). Instead of all keys expiring at exactly their TTL, each request computes a probability of refreshing the cache before expiration. As the TTL approaches zero, the probability increases. In practice, some request will refresh the key slightly before it expires, so the cache never actually goes empty.

3. Staggered TTL with jitter. Set TTLs as base_ttl + random(0, jitter_range). If 10,000 keys all have a 60-second TTL, they all expire at the same second. With jitter of 10 seconds, they expire spread across a 10-second window. This does not prevent stampede on a single key, but it prevents cache avalanche across many keys.

When to Choose Which

Choose Memcached when your workload is simple key-value caching with high concurrency, you do not need persistence, and you want predictable memory behavior. Memcached shines as a pure caching layer in front of a relational database where the application handles all data structure logic.

Choose Redis for everything else. If you need sorted sets for leaderboards, pub/sub for real-time features, streams for event logs, persistence for session stores, or Lua scripts for atomic multi-step operations, Redis is the answer. Its ecosystem is larger, its feature set is richer, and it can serve as both cache and lightweight data store.

In practice, Redis has become the default choice for most new systems. Memcached retains a strong position in legacy architectures and in environments where its multi-threaded performance on simple operations justifies the feature trade-off.

Further Reading

• Redis Benchmark Documentation, redis.io
• Memcached vs. Redis: Performance at Scale, AWS Whitepapers
• Thundering Herd Problem, Wikipedia
• A Crash Course in Caching (Final Part), ByteByteGo
• Redis vs Memcached in 2025, ScaleGrid

When "Big" Stopped Being an Adjective

For most of this course, we have discussed systems that handle thousands or millions of requests. Big data is what happens when the data itself becomes the bottleneck, not because the queries are complex, but because there is simply too much of it for any single machine to store, process, or analyze in a reasonable time.

The global datasphere hit roughly 64 zettabytes in 2020. By 2025, that number reached approximately 181 zettabytes. Projections for 2028 exceed 390 zettabytes. A single zettabyte is one trillion gigabytes. This growth is not linear. It is exponential, driven by IoT sensors, video streaming, mobile devices, and machine-generated logs.

This session introduces the fundamental processing paradigms and storage architectures that handle data at this scale.

OLTP vs. OLAP

Before diving into big data architectures, you need to understand the two fundamental database workload types. Every system leans toward one or the other.

Batch vs. Stream Processing

Data processing at scale splits into two paradigms based on when the data is processed relative to when it arrives.

Batch processing collects data over a period (hours, a day), then processes it all at once. MapReduce and Apache Spark are the canonical tools. You run a job that reads a full dataset, transforms it, and writes results. The job might take minutes or hours. The output is complete and consistent, but it is always behind real time by at least the batch interval.

Stream processing processes data as it arrives, record by record or in micro-batches of milliseconds. Apache Flink, Kafka Streams, and Apache Spark Structured Streaming are the primary tools. Latency is seconds or less. But handling out-of-order events, late arrivals, and exactly-once semantics is significantly more complex.

Most production systems use both. The Lambda architecture runs a batch layer for accurate historical data and a speed layer for approximate real-time results. The Kappa architecture simplifies this by using a single stream processing layer for everything, replaying the event log when historical reprocessing is needed.

Data Lake vs. Data Warehouse vs. Data Lakehouse

Where you store data at scale depends on its structure, who needs it, and how it will be queried.

A data lake is cheap storage for everything. Raw files in any format (JSON, CSV, Parquet, images, logs) dumped into object storage like S3 or Azure Data Lake Storage. Schema is applied when reading ("schema-on-read"), not when writing. This makes ingestion fast and flexible, but querying is slow without additional tooling. Data lakes tend to become "data swamps" without governance.

A data warehouse is structured storage optimized for analytical queries. Data is cleaned, transformed, and loaded into a predefined schema ("schema-on-write"). Columnar storage formats enable fast aggregations. Snowflake, BigQuery, and Redshift are the dominant players. Warehouses are fast for BI queries but expensive for raw storage and inflexible for unstructured data.

A data lakehouse combines both. It adds a transaction layer (Delta Lake, Apache Iceberg, Apache Hudi) on top of data lake storage. This gives you ACID transactions, schema enforcement, and query optimization on cheap object storage. You get warehouse-like performance without copying data into a separate warehouse system.

Kafka as the Integration Backbone

Apache Kafka sits at the center of most modern data pipelines. It is a distributed event streaming platform that acts as a durable, high-throughput buffer between data producers and consumers. Kafka does not process data. It transports it reliably and lets multiple downstream systems consume the same events independently.

A Kafka topic is an ordered, append-only log of events. Producers write to topics. Consumers read from topics at their own pace. Because the log is persistent (stored on disk with configurable retention), a new consumer can start reading from the beginning. A real-time dashboard can read events as they arrive. A batch pipeline can read the same events hours later. Neither interferes with the other.

This decoupling is why Kafka is the backbone. It separates "when data is produced" from "when and how data is consumed," which lets teams build and evolve their processing pipelines independently.

A Modern Data Pipeline

Putting these pieces together, here is what a modern data architecture looks like for a company that needs both real-time insights and historical analytics.

Data flows left to right. All sources publish to Kafka. Stream processors consume from Kafka for real-time needs. Batch processors consume the same data (or read from the lakehouse) for historical analysis. The lakehouse serves as the central storage layer, feeding both the warehouse for BI and ML pipelines for model training.

The critical insight: Kafka decouples producers from consumers. Adding a new downstream system (a fraud detection engine, a recommendation model, a compliance audit) does not require changing any upstream source. You just add a new consumer group to the relevant Kafka topic.

This second diagram shows a common pattern: Change Data Capture (CDC) using Debezium to stream database changes into Kafka, which then feeds into the analytics layer. This lets you keep your OLTP database optimized for transactions while building your OLAP layer from the same data, without any direct coupling between the two systems.

Further Reading

• What is a Data Lakehouse?, Databricks
• Data Generation Volume Worldwide 2010-2029, Statista
• Apache Kafka Documentation, apache.org
• Data Warehouse vs Data Lake vs Data Lakehouse, Monte Carlo
• Apache Flink Documentation, apache.org