Book summary “Designing Data-Intensive Applications”: summary

Introduction

You live in an era where applications are rarely bounded by the raw processing power of a single central processing unit. Your primary engineering challenges instead stem from the sheer volume of information you must store and the rapid pace at which your datasets grow. Compute-intensive workloads require clever algorithms, but data-intensive applications require you to master entirely different constraints. You must contend with the physical realities of unreliable networks, faulty hardware, and the demands of concurrent users who expect instantaneous responses.

Martin Kleppmann argues that building robust distributed systems requires you to balance three foundational pillars: reliability, scalability, and maintainability. When you fail to account for these trade-offs early in your architecture, your systems will eventually buckle under scale. You cannot simply buy a faster server or adopt a trendy database to solve fundamental architectural mismatches. You need a rigorous mental model of how data systems store, move, and process information.

This book provides a comprehensive tour of the fundamental principles underlying modern data infrastructure. You will explore how data models shape application code, how storage engines organize bytes on disk, and how encoding formats allow you to evolve schemas over time without breaking clients. You will examine the complex mechanics of replication, partitioning, and distributed transactions that make scaling possible. You will learn to navigate the inherent unreliability of networks and clocks, and you will discover how batch and stream processing architectures transform raw inputs into dependable derived data.

To design resilient architectures for your organization, you must be able to answer several critical design questions.

• How do you choose the right data model and storage engine for your specific workload patterns?
• What mechanisms ensure your distributed database maintains consistency despite network partitions and node failures?
• How do you manage schema evolution when multiple services read and write shared data concurrently?
• When should you process data in real-time streams rather than traditional batch pipelines?
• How do you balance the trade-offs between availability and consistency in distributed consensus protocols?

1. Trade-Offs in Data Systems Architecture

An application is classified as data-intensive when data management serves as its primary development challenge, distinct from compute-intensive systems that focus on parallelizing huge computations. You will find that these applications worry about storing and processing large data volumes, managing data changes, and guaranteeing consistency despite failures.

An application is classified as data-intensive when data management serves as its primary development challenge, requiring careful choices across relational, analytical, and distributed building blocks.

You rely on standard building blocks such as databases, caches, search indexes, stream processing, and batch processing to solve these challenges.

Operational Versus Analytical Systems

Data systems are fundamentally split between operational systems for user-facing interactive transactions and analytical systems optimized for reporting and data exploration.

• Online transaction processing (OLTP) handles low-latency point queries and individual record mutations for interactive applications.
• Online analytical processing (OLAP) scans massive record sets to calculate aggregate statistics for business intelligence or detect fraud patterns using ad-hoc queries.

Organizations separate analytical workloads from operational databases by routing data through extract-transform-load (ETL) pipelines into centralized data warehouses or flexible data lakes. A data warehouse stores a read-only, transformed copy of data extracted from multiple OLTP systems to support business reporting without impacting operational performance. A data lake holds raw files of any format, prioritizing flexibility and cost-effectiveness for data science tasks like feature engineering and machine learning model training.

Systems of Record and Derived Data

System architectures gain clarity by distinguishing between primary sources of truth and redundant, derived data systems optimized for read performance.

A system of record holds the authoritative, normalized version of data where each fact exists exactly once. Derived data systems take existing data and transform or denormalize it into caches, indexes, materialized views, or machine learning models that can be recreated if lost.

Distributed Versus Single-Node Systems

While distributed systems provide fault tolerance, scalability, and geographic distribution, they introduce network failure modes, latency, and operational complexity. Distributed systems communicate via networks where requests can timeout or fail ambiguously. Because single-node execution is often significantly simpler and cheaper, you should avoid premature distribution unless required by application constraints. Modern data system architecture is also legally and ethically constrained by privacy regulations, data minimization principles, and societal impacts. Regulations like the General Data Protection Regulation (GDPR) introduced in 2018 mandate explicit collection purposes, limited retention, and user rights such as the right to be forgotten.

2. Defining Nonfunctional Requirements

Application requirements are split into functional requirements, describing what the software must do, and nonfunctional requirements, describing qualities like performance, reliability, and maintainability. While functional requirements dictate features like buttons and screens, nonfunctional requirements determine overall success because an unbearably slow or unreliable app fails its purpose.

Nonfunctional requirements determine overall success, demanding rigorous handling of response time percentiles, hardware faults, and sociotechnical maintainability pillars.

Describing Performance

When you evaluate a system, load must be described with parameters like the rate of requests or the ratio of reads to writes. Scalability describes a system's ability to cope with increased load, often achieved through shared-nothing architectures. While vertical scaling and shared-disk architectures face cost and locking bottlenecks, horizontal scaling distributes independent nodes over a network to achieve linear growth potential.

Using averages for response times hides performance outliers, making high percentiles essential for evaluating user experience. Because response times vary due to background tasks or queueing delays, reporting the median or higher percentiles like p95, p99, and p999 reveals tail latencies experienced by real users.

Amazon measures response time requirements for internal services at the 99.9th percentile because customers with the most data and purchase history have the slowest requests.

Reliability and Fault Tolerance

Large-scale systems treat hardware faults as routine occurrences and rely on redundancy to maintain reliability. Because components fail at predictable annual rates across massive datacenters, systems add component redundancy and software-level fault tolerance.

Magnetic hard drives fail at an annual rate of 2% to 5%, meaning a storage cluster of 10,000 disks expects roughly one disk failure per day. Beyond that, 0.5% to 1% of solid state drives fail per year, 1 in 1,000 machines has a CPU core computing wrong results, and more than 1% of machines encounter uncorrectable RAM errors annually.

Maintainability

Maintainability requires designing software to minimize the long-term costs of bug fixing, adaptation, and technical debt. Software maintenance is managed through three core principles:

• operability for smooth operations;
• simplicity for reducing complexity;
• evolvability for adapting to future requirements.

In the UK Post Office Horizon scandal, accounting software bugs caused unhandled shortfalls that led to wrongful convictions, bankruptcies, and imprisonments over a twenty-year period spanning from 1999 to 2019.

3. Data Models and Query Languages

Data models shape how you write software and how you think about problems. Choosing the right model dictates whether queries remain fast and code stays clean as data grows.

Choosing between relational, document, and graph data models involves balancing impedance mismatch, data locality, and schema flexibility.

Relational Versus Document Models

Storing object-oriented application code data in relational tables creates an impedance mismatch that requires an awkward translation layer. Object-relational mapping frameworks reduce boilerplate code but introduce issues like the N+1 query problem, where an application performs N separate database lookups for related records instead of executing a single join.

Displaying a user profile requires fetching nested positions and education records.

The document data model organizes related data into self-contained tree structures using JSON, offering better read and write locality for one-to-many relationships. A LinkedIn profile containing positions and contact information inside user ID 251 lives within a single JSON document.

Normalization, Denormalization, and Joins

Normalized data models store human-meaningful information in a single place using IDs to avoid redundancy, while denormalized models duplicate information for faster reads at the cost of update complexity. You must choose between fast writes with complex joins or fast reads with duplicated data.

• Normalized representations prevent inconsistencies and speed up writes;
• denormalized representations speed up reads through duplicated data;
• join tables handle many-to-many relationships in relational systems.

Storing a region as an ID referencing a standardized regions table keeps data clean. Storing the region name as a plain-text string directly in every user profile speeds up reads.

Graph-Like Data Models

Property graphs model complex many-to-many relationships using vertices and directed edges, allowing efficient multi-hop graph traversals in both forward and backward directions.

Modeling people and locations involves vertices like Lucy from Idaho and Alain from France connected by edges such as BORN_IN, LIVING_IN, and WITHIN. Query languages like Cypher and SPARQL enable concise pattern matching and variable-length path traversals that require complex recursive common table expressions in SQL. A query that takes only 4 lines in Cypher can require a 31 lines SQL recursive common table expression.

Declarative query languages specify what data to retrieve rather than how to execute the retrieval, allowing database query optimizers to improve performance automatically. You write what you need using SQL or Cypher, and the database executes the query efficiently across multiple CPU cores.

4. Storage and Retrieval

OLTP storage engines balance write throughput and read performance through log-structured merge-trees and B-trees, while analytics rely on column-oriented layouts.

Storage and Indexing for OLTP

Log-structured storage engines use append-only files and in-memory memtables flushed to immutable Sorted Strings Tables to achieve high write throughput. Writes are first appended to an unsorted durability log and inserted into an in-memory ordered data structure called a memtable. When the memtable exceeds a threshold, it is written to disk as an immutable SSTable segment. Background merging and compaction processes combine segments and discard overwritten or deleted values using tombstones.

The algorithm was originally published in 1996 under the name Log-Structured Merge-tree, also known as an LSM-tree. RocksDB, Cassandra, ScyllaDB, and HBase use this algorithm, which was inspired by Google's Bigtable paper.

B-Trees

B-trees break the database into fixed-size pages and update data in place, serving as the standard index implementation in relational databases. B-trees maintain key-value pairs in sorted order across fixed-size pages, typically 4 KiB to 16 KiB. To ensure crash resilience and prevent torn pages, B-tree implementations rely on a write-ahead log where modifications are recorded before being applied to the tree pages.

PostgreSQL uses 8 KiB pages and MySQL uses 16 KiB pages by default. B-trees were introduced in 1970. A four-level tree with 4 KiB pages and a branching factor of 500 can reach a maximum storage capacity of 250 TB.

Write amplification measures the total ratio of bytes written to disk versus the logical bytes written by the application, directly impacting storage longevity and throughput. Because storage engines perform background compactions, log appends, and page writes, a single application write translates into multiple disk I/O operations. Lower write amplification helps preserve SSD lifespan and maintains higher write throughput under heavy workloads.

Data Storage for Analytics

Column-oriented storage stores values from each column together rather than grouping an entire row's values, dramatically improving analytical query performance. Analytical queries typically access only a small subset of a table's columns. Columnar layouts allow the query engine to read and parse exclusively the required columns from disk, which pairs exceptionally well with heavy data compression techniques like bitmap encoding and run-length encoding.

Parquet is a columnar storage format supporting document data models using shredding or striping techniques based on Google's Dremel.

Analytical query engines use compilation or vectorization to minimize CPU overhead and efficiently process millions of rows. Instead of interpreting rows one by one through general-purpose operators, engines either generate and compile machine code via JIT compilation or process batches of column values through predefined vector operators.

Multidimensional indexes and inverted indexes enable efficient querying across multiple attributes or text terms simultaneously. Specialized spatial structures like R-trees and Bkd-trees handle multi-attribute range queries such as geospatial coordinates. For full-text search, inverted indexes map terms to postings lists of document IDs, allowing rapid keyword and substring lookups using bitwise operations.

PostGIS implements geospatial indexes as R-trees using PostgreSQL's Generalized Search Tree indexing facility. Lucene uses inverted indexes and Levenshtein automata for fast fuzzy matching, where edit distance represents one letter added, removed, or replaced in fuzzy search.

5. Encoding and Evolution

Applications change constantly. You must modify features, business logic, and data storage formats. To support rolling upgrades without downtime, applications require both backward and forward data compatibility.

Supporting rolling upgrades requires backward and forward data compatibility via compact binary formats like Protocol Buffers and Apache Avro.

Staged rollouts mean old and new code versions coexist in production. Newer code must read older data, while older code must read newer data. Language-built-in serialization libraries tie data formats to a single platform and expose security vulnerabilities. Decoding untrusted byte sequences can instantiate arbitrary classes and enable remote code execution.

JSON, XML, and Binary Variants

Textual formats like JSON and XML offer human readability but introduce specific trade-offs. JSON does not distinguish integers from floating-point numbers. JavaScript uses IEEE 754 double-precision floating-point numbers, meaning integers greater than $2^{53}$ lose precision.

X uses a 64-bit number to identify each post. Its JSON API returns post IDs twice, once as a JSON number and once as a decimal string, to prevent rounding errors.

Protocol Buffers

Protocol Buffers and Apache Thrift require a schema defining fields and numeric field tags. Encoded records omit field names, packing type annotations and tag numbers into compact bytes using variable-length integers. Protocol Buffers required 33 bytes to encode the sample record.

Field names can change freely during schema evolution. Field tags cannot be altered because existing data relies on them.

Avro

Apache Avro, started in 2009 as a Hadoop subproject, encodes data compactly as a concatenation of values without field names or types. Avro required 32 bytes to encode the sample record, making it the most compact format.

Avro relies on separate writer and reader schemas:

• writer's schema used when data is generated;
• reader's schema expected by the currently running code version.

Schema resolution compares both schemas, matching fields by name and filling in default values for missing attributes.

6. Replication

Replication distributes data across multiple machines via a network. You use replication to keep data geographically close to users, maintain availability if nodes fail, and scale read throughput.

Replication distributes data across nodes using single-leader, multi-leader, or leaderless configurations, each introducing distinct consistency and conflict resolution trade-offs.

Single-Leader Replication

Every write must go through a designated leader node. The leader applies the write locally and sends it to followers as part of a replication log or change stream. Followers apply these changes in the identical order to match the leader's dataset.

• PostgreSQL, MySQL, and MongoDB utilize single-leader replication;
• read replicas and hot standbys serve client read queries.

Synchronous replication makes the leader block and wait for a follower confirmation. Asynchronous replication returns success immediately without waiting for slow or failed followers.

Asynchronous replication risks lost writes if the leader crashes before replication occurs. You typically configure one synchronous follower while other followers operate asynchronously to maintain performance and durability.

Problems with Replication Lag

Asynchronous replication introduces replication lag. If you query an asynchronous follower, you might read stale data that has not caught up with the leader.

Read-after-write consistency ensures that you always see your own updates. Monotonic reads prevent you from seeing time move backward when your queries hit different replicas randomly. Consistent prefix reads ensure you witness answers in the correct causal order.

Handling node outages requires detecting failures using timeouts and promoting a follower to leader via failover. If asynchronous replication was used, the old leader's unpropagated writes get dropped permanently.

GitHub experienced an incident where an out-of-date MySQL follower was promoted to leader. It reused an autoincrementing primary key that had already been assigned in Redis, causing private data disclosure.

Leaderless Replication

Leaderless Dynamo-style databases allow any replica to accept writes directly. Clients send writes and reads to multiple replicas in parallel.

• Riak, Cassandra, and ScyllaDB use leaderless architectures;
• 3 or 5 represents a typical odd number of replicas in a quorum configuration.

By enforcing that the write quorum plus read quorum exceeds the total replica count, readers intersect with at least one up-to-date replica. Systems repair stale data using read repair, hinted handoff, and anti-entropy background processes.

Concurrent writes in multi-leader and leaderless systems cause write conflicts that require resolution. Systems can use last-write-wins to discard concurrent writes, provide manual conflict resolution, or apply automatic conflict resolution algorithms.

Amazon's shopping cart anomaly demonstrated that merging concurrent cart updates by taking a set union can cause deleted items to reappear. Systems determine if operations are concurrent by tracking causal dependencies using version vectors instead of physical clocks.

7. Sharding

Datasets that grow too large for a single machine require sharding. You partition data and write throughput across multiple nodes so that each record belongs to exactly one shard.

Sharding partitions large datasets across nodes using key ranges or hashes, requiring careful handling of secondary indexes and rebalancing operations.

Databases typically combine sharding and replication. Each shard acts as a small database, and replication maintains follower copies of each shard on different nodes for fault tolerance.

In multitenant SaaS and cloud services, storing separate tenant datasets in distinct shards provides resource and permission isolation. Sharding for multitenancy prevents a computationally expensive operation by one tenant from affecting others while simplifying data residence requirements.

Sharding by Key Range

Key-range sharding assigns contiguous ranges of partition keys to each shard. This design enables efficient range queries because keys are sorted in structures like B-trees or SSTables.

However, monotonically increasing keys create a risk of write amplification hot spots. If your partition key is a timestamp, all writes overload the current time range shard. Systems like HBase address default region split sizes by dynamically managing boundaries, but you must still consider hot keys.

Sharding by Hash of Key

Hash-range sharding applies a hash function to keys before mapping them to shards. This technique uniformly distributes skewed input data across a range of hash values, eliminating hot spots.

You sacrifice efficient key-range scans when using hashes. Systems minimize data movement during cluster rebalancing by adopting strategies such as:

• consistent hashing;
• rendezvous hashing;
• jump consistent hashing.

Default configurations dictate these boundaries. Cassandra uses 16 hash ranges per node by default, while ScyllaDB uses 256 hash ranges per node.

Global Secondary Indexes

Local secondary indexes are maintained independently by each shard for its own data. A write only updates the specific shard containing the record, but searching across secondary index values requires querying every shard in parallel and causes tail latency amplification.

Global secondary indexes cover data across all shards and are themselves term-partitioned by the indexed value. This approach optimizes reads for single terms because queries read from only a single index shard. Writing a single record can affect multiple index shards, which requires distributed transactions or asynchronous updates.

Distributed databases route queries to the correct shard replica using clients, routing tiers, or random nodes with request forwarding. Systems like HBase and SolrCloud rely on ZooKeeper and consensus protocols to track authoritative mappings.

8. Transactions

Transactions provide a logical grouping of reads and writes that either completely succeed or fail atomically, simplifying application error handling.

The ACID acronym stands for atomicity, consistency, isolation, and durability. Atomicity ensures all-or-nothing abortability on failure. Consistency preserves application invariants via constraints. Isolation prevents concurrent interference. Durability guarantees committed data is not lost.

The Post Office Horizon scandal was technically caused by a lack of proper ACID transactions in the underlying accounting system.

Read-committed isolation provides basic concurrency protection by preventing dirty reads and dirty writes. Preventing dirty reads ensures uncommitted writes are invisible. Preventing dirty writes guarantees that concurrent updates to the same row do not intermingle.

Aaliyah and Bryce simultaneously attempt to buy the same car on a used car sales website. Read-committed prevents mixing up which buyer gets the listing and the invoice.

Snapshot Isolation and Repeatable Read

Snapshot isolation ensures each transaction reads from a consistent database snapshot using multiversion concurrency control to prevent read skew.

Readers never block writers and writers never block readers. Visibility rules determine which row versions are observable based on transaction start times and active in-progress transaction lists. PostgreSQL transaction IDs are 32-bit integers which overflow after approximately 4 billion transactions.

Aaliyah splits $1,000 across two accounts and transfers $100. Under read-committed she might observe read skew and see $900 total momentarily, which snapshot isolation avoids.

Lost updates happen during concurrent read-modify-write cycles and can be prevented using atomic write operations, explicit locking, or automatic detection. When two clients read a value, modify it locally, and write it back, one write can clobber the other. Databases prevent this via atomic instructions, exclusive row locks using clauses like FOR UPDATE, or automatic detection mechanisms that abort conflicting transactions.

Two concurrent clients increment a counter stored in a database, reading 42 and both writing 43 instead of 44.

Write Skew and Phantoms

Write skew occurs when concurrent transactions read overlapping data sets, make decisions based on valid premises, and write conflicting updates that violate invariants.

Unlike lost updates or dirty writes, transactions in write skew modify different objects. When queries check for the absence of rows matching a condition and then insert matching rows, phantoms occur. Snapshot isolation does not prevent write skew unless explicit locks or serializable isolation are used.

Aaliyah and Bryce are the on-call doctors for a shift. Both feel sick simultaneously, check that 2 doctors are on call, and take themselves off call, leaving zero doctors on duty.

Two-phase locking is a pessimistic concurrency control algorithm that enforces serializability by blocking readers and writers mutually. Writers block readers and readers block writers using shared and exclusive locks held until the end of the transaction. Two-phase locking was virtually the only widely used algorithm for database serializability for approximately 30 years.

A transaction reading an entire table for a backup takes a shared table lock, blocking all concurrent write transactions until it commits. Serializable snapshot isolation is an optimistic concurrency algorithm that detects serialization conflicts on top of multiversion concurrency control without blocking readers. First described in academic literature in 2008, serializable snapshot isolation allows transactions to proceed without locks.

Two-Phase Commit

Two-phase commit is an atomic commitment protocol used to ensure that all participants in a distributed transaction either successfully commit or abort together.

Coordinated by a transaction manager, two-phase commit uses a prepare phase where participants promise to commit if requested. This is followed by a commit phase after writing the decision to disk. If the coordinator crashes while participants are in-doubt, participants must wait for recovery or risk inconsistency.

Database nodes receive commit requests, but some crash or timeout while others succeed, leading to atomicity failures without an explicit atomic commit protocol.

9. The Trouble with Distributed Systems

Unlike single computers that are either fully functional or completely crashed, distributed systems experience partial failures where some parts break unpredictably while others keep working. In a distributed system, network and multi-node operations are nondeterministic, meaning they may sometimes succeed and sometimes fail unpredictably. Tolerating these partial failures allows you to perform rolling upgrades and build reliable systems from unreliable components.

Coda Hale noted dealing with long-lived network partitions, PDU failures, whole-DC power outages, and a hypoglycemic driver crashing a Ford pickup truck into a datacenter HVAC system.

Unreliable Networks

While TCP provides reliable byte-stream delivery with congestion and flow control, it cannot eliminate network unreliability because it cannot distinguish between lost outbound packets, lost acknowledgments, or down remote nodes. A medium-sized datacenter study found 12 network faults per month, split evenly between single-machine and entire-rack disconnections.

Packet delay variability in computer networks is predominantly caused by queueing on congested network links, destination machine OS buffers, and virtual machine monitors.

Unreliable Clocks

Modern computers rely on time-of-day clocks for wall-clock time and monotonic clocks for measuring elapsed time, but both types suffer from hardware drift, synchronization flaws, and jump anomalies. Time-of-day clocks synchronize via Network Time Protocol (NTP) but can jump backward or forward when reset, making them dangerous for measuring durations or ordering events. Monotonic clocks are safe for measuring intervals locally but cannot be compared across different machines.

Google assumes a 200 ppm clock drift for its servers, equivalent to a 6 ms drift over 30 seconds or a 17-second drift over a day. Experiments also observed a 35 ms minimum internet synchronization error.

Distributed nodes can experience arbitrary execution pauses due to garbage collection, VM live migration, CPU contention, or disk I/O, causing locks and leases to expire silently while a paused node continues operating. When a leader node is paused, other nodes declare it dead and elect a new leader.

Knowledge, Truth, and Lies

Distributed systems must cope with partial failures, unreliable packet networks, drifting clocks, and process pauses using majority quorums and fencing tokens.

A single node cannot trust its own state or judgment because it might be partitioned or paused. Requiring a majority quorum ensures that decisions like declaring a node dead are unambiguous. Fencing tokens, which are monotonically increasing numbers issued alongside leases, allow storage services to reject writes from outdated zombie nodes using numbers such as 33 and 34.

HBase suffered from data corruption bugs due to expired leases until fencing mechanisms were properly enforced.

Formalizing system behavior requires defining system models with specific timing and fault assumptions, and categorizing correctness guarantees into safety and liveness properties. The partially synchronous model with crash-recovery faults is the most realistic abstraction for distributed systems.

• Safety properties dictate that nothing bad happens and must always hold.
• Liveness properties state that something good eventually happens under certain recovery assumptions.

10. Consistency and Consensus

Linearizability guarantees that a replicated database behaves as if there is only a single copy of the data, providing an immediate recency guarantee where reads always return the most up-to-date value. It ensures that once a value has been written or read, all subsequent reads see that value or a newer one.

Linearizability requires that operations take effect atomically at a definite point in time, moving forward without reversing.

Aaliyah excitedly tells Bryce the final sports score she just saw on her phone, but when Bryce immediately reloads his phone, a lagging replica shows the game is still ongoing, violating linearizability.

Serializability and linearizability are distinct guarantees. Serializability isolates multi-object transactions into a serial execution order, whereas linearizability provides a single-object recency guarantee. Serializability allows stale reads as long as transactions do not interleave. Linearizability requires that if one operation finishes before another starts, the later operation must observe a state at least as new.

Linearizability

Linearizability is essential for implementing distributed locks, leader election leases, hard uniqueness constraints, and preventing cross-channel timing race conditions.

A web server writes an uploaded video to file storage and then uses a message queue to tell a transcoder to process it. If file storage is not linearizable, the transcoder might fetch a stale version.

The CAP theorem states that an application requiring linearizability under network partitions must become unavailable on disconnected replicas, forcing a choice between consistency and availability. Google data shows that network partitions cause 8% of incidents. If replicas are disconnected due to a network partition, a CP system must wait or return an error to preserve consistency, whereas an AP system allows independent processing at the cost of linearizability.

Consensus

Single-value consensus, linearizable compare-and-set operations, shared logs, atomic transaction commits, and fetch-and-add operations are all fundamentally equivalent problems. An algorithm that solves any one of these problems can be converted into a solution for any of the others.

These problems have distinct consensus numbers:

• fetch-and-add has a consensus number of 2;
• compare-and-set has a consensus number of infinity;
• shared logs have a consensus number of infinity.

Consensus in Practice

Shared logs act as a consensus abstraction that enables fault-tolerant state machine replication and serializable transaction execution. By processing log entries representing writes or deterministic transactions in the same order across all replicas, systems achieve consistent states.

Distributed consensus algorithms use epoch numbers and multi-round quorum voting to elect unique leaders and safely append log entries without split-brain scenarios. Nodes hold an election vote using higher epoch numbers when a leader is suspected dead. Before appending an entry, the leader collects quorum votes to verify no higher-numbered epoch leader has emerged.

Fault tolerance requires specific node counts:

• 3 nodes tolerate one failure;
• 5 nodes tolerate two failures.

11. Batch Processing

Batch processing jobs process large immutable datasets to produce derived outputs from scratch without mutating source data, enabling human fault tolerance through code rollbacks.

Batch processing frameworks leverage immutable datasets, distributed filesystems, and the shuffle algorithm to derive reliable outputs from scratch.

Unlike interactive online systems focused on response time, offline systems take read-only inputs. Treating inputs as immutable avoids side effects. If a bug corrupts the output, you can simply roll back the code or switch back to a previous output directory. Martin Kleppmann calls this human fault tolerance and time travel.

Distributed Filesystems and Object Stores

Distributed batch frameworks rely on distributed filesystems and object stores that break files into large blocks and use replication or erasure coding to handle commodity hardware failures. Distributed filesystems like HDFS break files into large blocks distributed across many machines, using a NameNode to track metadata and DataNodes to expose local filesystems. Object storage services use bucket and key structures where objects are immutable and require full rewrites, decoupling storage from compute to allow independent scaling.

The shared-nothing principle guides this hardware layout:

• HDFS uses a default block size of 128 MB;
• JuiceFS and object stores use a block size of 4 MB;
• ext4 uses a standard block size of 4,096 bytes.

Batch Processing Models

Modern dataflow engines have largely replaced MapReduce by processing entire workflows as single jobs with flexible operator combinations and in-memory intermediate state. Google published MapReduce in 2004, breaking data processing into alternating mapper and reducer phases separated by forced file sorting and distributed filesystem writes. Dataflow engines like Spark and Flink treat workflows as directed graphs, combining adjacent operators and executing tasks as soon as their inputs are ready.

Because batch jobs regenerate outputs from scratch, a task failure can be handled by deleting partial outputs and rescheduling the task on another machine rather than rerunning the whole job. Companies reduce costs by running low-priority tasks on spare capacity. You can use AWS spot instances, spot virtual machines, or GCP preemptible instances, which can be killed at any moment by the scheduler during preemption.

Shuffling Data

The shuffle is a distributed sorting algorithm used by batch processors to shard, sort, and route key-value pairs between mappers and reducers for joins and aggregations. In a shuffle, mappers write sorted key-value pairs into separate local files for each reducer based on a hash of the key. Reducers copy these files from all mappers, perform a mergesort-style merge to align identical keys consecutively, and then invoke the reduce function on each key's collection of values.

Distributed job orchestrators manage batch execution by combining a resource manager, centralized schedulers, and node executors with performance isolation. Orchestrators such as Hadoop YARN and Kubernetes act like distributed operating systems kernels. Task executors run on each node to handle job tasks and enforce security and performance isolation using Linux cgroups. Centralized resource managers maintain cluster state using tools like ZooKeeper or etcd and use heuristics to allocate resources.

12. Stream Processing

Batch processing requires a complete input set because operations like sorting need the final record to determine the first output. In contrast, stream processing abandons fixed time slices to handle unbounded data as it arrives incrementally over time, reducing latency for impatient users.

Stream processing handles unbounded data through log-based brokers, change data capture, event time windows, and exact-once fault tolerance semantics.

Transmitting Event Streams

AMQP and JMS-style message brokers delete messages after consumption, making processing destructive. Log-based message brokers like Apache Kafka append messages to an immutable log stored on disk, allowing multiple consumers to read independently without deleting records, keeping consumer offsets to track progress.

A typical large hard drive provides 20 TB capacity and achieves 250 MB/s sequential write throughput.

Databases and Streams

Dual writes can cause race conditions and permanent inconsistency when interleaved concurrently across different systems. Change data capture observes all writes written to a database and extracts them as an ordered change stream, allowing derived data systems like search indexes and caches to stay synchronized without dual writes.

Rather than treating databases as primary storage, event sourcing and changelogs consider the append-only log as the system of record. Read-optimized views and current states are derived from this log, making recovery from bugs and auditing much simpler.

Reasoning About Time

Relying on local processing time instead of event timestamps introduces severe artifacts and bad data during network delays, system restarts, or backlog processing. Stream processors must distinguish between event time when the action occurred and processing time when the system handles it, managing straggler events either by ignoring them or by publishing corrections.

Stream processing frameworks utilize different window types to bound and aggregate unbounded event streams:

• tumbling windows have fixed, non-overlapping lengths;
• hopping windows have fixed lengths with overlaps for smoothing;
• sliding windows cover events occurring within a specified interval of each other;
• session windows group user events based on inactivity periods.

Achieving exactly-once semantics in stream processing requires mechanisms like microbatching, checkpointing, or idempotent writes to discard partial outputs from failed tasks without losing progress. Because streams never end, restarting failed tasks from the beginning is impossible, requiring stream processors to use rolling checkpoints to capture state and offset markers.

13. A Philosophy of Streaming Systems

Complex applications require multiple specialized storage tools like OLTP databases, search indexes, and caches. Federated databases provide a unified read query interface across disparate storage technologies. Synchronizing writes across these heterogeneous components remains a much harder engineering problem.

Unbundling database architectures around asynchronous event logs, end-to-end request IDs, and derived data achieves superior resilience and scalability.

Following the Unix philosophy, the unbundled database approach uses small tools communicating through uniform APIs. Change Data Capture and event sourcing allow you to synchronize writes using asynchronous event logs instead of fragile distributed transactions.

• Derived data is updated asynchronously through deterministic retries.
• Polystores combine multiple engines to serve specialized read queries.

Data Integration

Distributed transactions use atomic commit protocols that suffer from poor fault tolerance and performance. Log-based derived data achieves correctness asynchronously while containing faults locally.

Constructing a totally ordered event log works for small systems. As you scale across shards, datacenters, microservices, and offline clients, total order becomes limited because consensus requires routing all events through a single leader node.

Unbundling Databases

Treating inputs as immutable and outputs as append-only lets you reprocess historical data through batch and stream processors. You can restructure datasets and evolve applications gradually without system downtime.

In 19th-century England, railway networks converted competing rail gauge standards by adding a third rail to create mixed gauge tracks. Trains of both gauges ran before workers removed the old rail.

Aiming for Correctness

Low-level protocol features cannot prevent duplicate operations if a client loses connection before receiving a confirmation and retries. You must pass a client-generated unique request identifier all the way to the database.

In 1984, Saltzer, Reed, and Clark articulated the end-to-end argument. Enforcing a uniqueness constraint in a requests table ensures you prevent duplicate money transfers during network retries.

Asynchronous dataflow systems decouple timeliness from integrity, permitting temporary stale reads. You gain massive scalability and robustness while protecting absolute data correctness without distributed transactions.

14. Doing the Right Thing

Data systems do not operate in a moral vacuum. When you build scalable architectures, your technical choices carry profound societal consequences. You must evaluate the human cost of predictive analytics, behavioral tracking, and data accumulation.

Architects must evaluate the societal impacts of predictive analytics, surveillance infrastructure, and data accumulation to prevent algorithmic harm.

Bias and Discrimination

Machine learning models trained on historical data often learn and amplify existing societal biases. Predictive analytics extrapolate from past patterns to encode historical discrimination. Inputs such as postal codes or IP addresses heavily correlate with protected traits like race in segregated neighborhoods, creating biased outputs.

Martin Kleppmann notes that machine learning is like money laundering for bias. Algorithms take prejudiced human decisions and sanitize them behind a veneer of mathematical neutrality.

• Organizations prioritize avoiding bad outcomes like defaults or untrustworthy hires.
• Automated systems label people as risky based on historical proxies.
• Individuals suffer repeated rejections across jobs, air travel, insurance, and housing.

This creates an algorithmic prison. Predictive analytics systems can systematically and arbitrarily exclude individuals from societal participation without proof of guilt or chance of appeal. Data-driven systems can also trigger pernicious self-reinforcing feedback loops that worsen individual misfortunes. A job loss from a bad credit score pushes someone further into poverty.

Economists found that when gas stations in Germany introduced algorithmic prices, competition was reduced and consumer prices went up because the algorithms learned to collude.

Surveillance

Behavioral tracking and data collection for ad-funded services have created an unprecedented mass surveillance infrastructure. When user activity is tracked as a side effect, services take on independent interests that conflict with users. Corporations amass intimate data about location, movements, and health to build detailed personal profiles.

True consent for data tracking is frequently absent because privacy policies are opaque and popular services are practically mandatory for social participation. You lack knowledge of how data is processed, and relationships with platforms are asymmetric. Network effects make popular services essential for modern life, meaning that opting out carries a heavy social and professional cost.

Privacy is not about keeping everything secret, but about maintaining the decision right to control what to reveal to whom. Surveillance infrastructure transfers the power to decide what is disclosed from you to the data collector. Companies keep profiling outcomes secret to avoid being perceived as creepy while maximizing profits.

Data as Assets and Power

Accumulated personal data acts as a hazardous material or toxic asset that carries severe long-term risks. Beyond marketing and monetization, stored data attracts government coercion, data brokers, and security breaches. Collecting large datasets creates long-term hazards if they fall into the hands of hostile actors or future repressive regimes.

Just as the Industrial Revolution brought severe pollution that required regulatory safeguards, the information age faces a data pollution problem. Unchecked data collection and retention fester over time, presenting environmental challenges for the information economy. Protecting privacy and practicing data minimization are necessary steps for sustainable technology development.

To synthesize these principles into your daily engineering practice, follow these core rules.

Careful workload analysis and forward-looking data encoding ensure your systems remain adaptable under changing operational demands.

1. Analyze your workload access patterns before choosing a storage engine, matching write-heavy systems to log-structured layouts and read-heavy transactional workloads to B-trees.
2. Design data encodings for forward and backward compatibility from the start, ensuring rolling upgrades do not break your communication protocols.
3. Accept the inevitability of partial failures in distributed environments, relying on majority quorums and consensus protocols instead of assuming network perfection.
4. Separate transactional processing from analytical processing to protect operational workloads from the heavy resource demands of ad-hoc queries.
5. Embrace immutable event logs and derived data systems as a resilient foundation for unbundling monolithic state management into decoupled services.

10 Key Ideas

1. Data systems are fundamentally split between operational systems for user-facing interactive transactions and analytical systems optimized for reporting and data exploration.
2. Supporting rolling upgrades requires backward and forward data compatibility via compact binary formats like Protocol Buffers and Apache Avro.
3. OLTP storage engines balance write throughput and read performance through log-structured merge-trees and B-trees, while analytics rely on column-oriented layouts.
4. Replication distributes data across nodes using single-leader, multi-leader, or leaderless configurations, each introducing distinct consistency and conflict resolution trade-offs.
5. Sharding partitions large datasets across nodes using key ranges or hashes, requiring careful handling of secondary indexes and rebalancing operations.
6. Transactions provide a logical grouping of reads and writes that either completely succeed or fail atomically, simplifying application error handling.
7. Distributed systems must cope with partial failures, unreliable packet networks, drifting clocks, and process pauses using majority quorums and fencing tokens.
8. Linearizability requires that operations take effect atomically at a definite point in time, moving forward without reversing.
9. Batch processing frameworks leverage immutable datasets, distributed filesystems, and the shuffle algorithm to derive reliable outputs from scratch.
10. Unbundling database architectures around asynchronous event logs, end-to-end request IDs, and derived data achieves superior resilience and scalability.