Distributed Systems Principles

Introduction

Big data systems are distributed because the volume, velocity, and diversity of data exceed the capacity of a single computer. Distribution provides storage capacity, processing power, and resilience, but it also introduces partial failure, network delay, coordination cost, and uncertainty. A distributed data system must therefore answer several fundamental questions: How is data divided? How are copies maintained? What happens when machines or networks fail? Which results are acceptable when replicas disagree? How can computation be placed near the data?

These questions are not implementation details. They determine the correctness, performance, cost, and operational behavior of systems such as distributed databases, cloud object stores, stream-processing platforms, data warehouses, and machine-learning infrastructures.

Learning Objectives

By the end of this chapter, you should be able to:

• explain partitioning, replication, parallelism, locality, and fault tolerance;
• distinguish consistency, availability, and partition tolerance;
• apply CAP and PACELC to distributed-system design choices;
• describe consensus and its role in coordination;
• classify failure models and select appropriate recovery strategies;
• evaluate a distributed architecture using performance, correctness, and governance criteria.

1. The Distributed-System Model

A distributed system is a collection of independent computing components that communicate through a network and appear to users as a coherent system. Components may include storage nodes, compute workers, coordinators, metadata services, message brokers, and clients.

Unlike a centralized system, a distributed system cannot assume that all components share a clock, memory, or failure domain. Communication takes time and messages may be delayed, duplicated, reordered, or lost. A node can fail while other nodes continue operating. Consequently, the system must make progress using incomplete information.

Two properties are especially important:

• Safety: nothing incorrect is returned or committed. For example, a transaction must not be acknowledged as committed if its required durability guarantee has not been achieved.
• Liveness: the system eventually makes progress. For example, a request should not wait forever when the required resources are available.

A design often trades some degree of liveness for safety during uncertainty. This is why a system may reject writes or become temporarily unavailable rather than risk conflicting decisions.

Important terminology

• Node: a machine or logical process participating in the system.
• Cluster: a group of nodes managed as one service.
• Coordinator: a component that directs an operation or manages metadata.
• Replica: a copy of data maintained on another node.
• Partition: a subset of a dataset assigned to a node or group of nodes.
• Quorum: a sufficient subset of participants whose responses establish an operation’s result.
• Partial failure: failure of some components while others remain operational.
• Network partition: a communication failure that separates components into groups that cannot reliably communicate.

2. Partitioning and Sharding

Partitioning divides a logical dataset into smaller subsets called partitions or shards. A partitioning function maps each record to a partition. If the key is k and the system has N partitions, a simple scheme is:

Here, h(k) is a hash value and p is the partition identifier. This expression is useful when partitions are fixed and evenly distributed. It becomes problematic when N changes, because many keys move. Consistent hashing, range partitioning, and virtual nodes reduce this disruption.

Common partitioning strategies

A good partition key distributes storage and requests while preserving the access patterns required by applications. A key that is unique but rarely queried may produce balanced storage yet poor query performance. Conversely, a popular key can create a hot partition, where one node receives disproportionate traffic.

Partitioning affects transactions. A transaction confined to one partition can often be processed locally. A transaction spanning several partitions requires coordination, such as two-phase commit or an application-level workflow, increasing latency and failure complexity.

Partitioning example

Suppose 1.2 billion events are divided across 12 equally sized partitions. The ideal average is:

This is only an average. If one customer generates 30 percent of all events and the customer identifier is the partition key, a single partition may become a bottleneck. Salting, composite keys, or time-based bucketing can distribute that workload, but these techniques may complicate reads and ordering.

3. Replication and Durability

Replication stores multiple copies of data. If the replication factor is r, each logical record has up to r physical copies. Replication supports availability, read scaling, and recovery from node failure, but it consumes storage and creates consistency obligations.

Replication may be:

• Synchronous: a write is acknowledged only after specified replicas confirm it. This provides a stronger durability point but adds latency and may reduce availability during failures.
• Asynchronous: the primary acknowledges before all replicas receive the write. This improves responsiveness but creates a replication lag and possible data loss if the primary fails before propagation.
• Semi-synchronous: the system waits for a selected set of replicas while allowing others to catch up later.

Replica placement matters. Three copies on one physical host do not provide the same protection as copies distributed across hosts, racks, availability zones, or regions. A placement policy should consider correlated failures, legal data-residency requirements, and network cost.

For a replicated partition with r = 3, losing one node may be harmless if two valid replicas remain. However, availability depends on the protocol, not only on the number of copies. A system may deliberately stop writes when it cannot establish the required quorum.

4. Parallelism, Locality, and Data Movement

Parallelism divides work among independent workers. In data-parallel processing, each worker processes a subset of records. If a job contains work amount W and uses P workers, the ideal execution time is approximately:

The formula assumes equal work, no communication, no startup overhead, and unlimited scalability. Real execution time is better represented as:

where t_overhead includes scheduling and startup costs, t_skew represents imbalance between workers, and t_communication represents network transfers and coordination.

If the serial fraction of a workload is s, Amdahl’s Law gives the theoretical speedup with P processors:

Here, 0 ≤ s ≤ 1. The model assumes a fixed-size workload and a clearly defined serial fraction. For s = 0.1 and P = 10,

Thus, ten workers cannot provide a tenfold speedup because ten percent of the work remains serial.

Data locality means placing computation near the data it reads. Moving computation is often cheaper than moving large datasets across a network. Locality can be storage-local, rack-local, zone-local, or regional. Modern cloud systems sometimes trade physical locality for elastic compute, using columnar formats, caching, compression, and data pruning to reduce movement.

A distributed join illustrates the issue. If both datasets are partitioned by the join key, corresponding records can be joined locally. If not, a shuffle redistributes records across the network. Shuffles may dominate runtime and increase failure-recovery cost.

5. Failure Models and Fault Tolerance

A failure model states what can go wrong and what assumptions the system makes. Common models include:

Timeouts require care. A timeout does not prove that a node has failed; the message may simply be delayed. Treating a slow node as failed can produce duplicate work or conflicting leadership unless operations are idempotent and fencing mechanisms are used.

Fault tolerance generally uses redundancy, detection, isolation, and recovery:

1. Redundancy: maintain replicas or alternative workers.
2. Detection: use heartbeats, leases, health checks, or client-observed failures.
3. Isolation: prevent a faulty component from corrupting new work.
4. Recovery: replay logs, rebuild replicas, reschedule tasks, or restore checkpoints.

A durable write-ahead log records intended changes before applying them to the main data structure. After a crash, the system replays committed log records and discards incomplete operations. Checkpointing periodically saves a recoverable state, reducing replay time.

6. Consistency and Availability

Consistency describes what values and ordering guarantees clients observe. It is not a single property. Examples include:

• Linearizability: each operation appears to occur atomically at one point between invocation and response, respecting real-time order.
• Sequential consistency: operations appear in one global order that preserves each client’s program order, but not necessarily real-time order.
• Causal consistency: causally related operations are observed in causal order; concurrent operations may be observed differently.
• Eventual consistency: if updates stop and communication continues, replicas eventually converge.
• Read-your-writes: a client sees its own successful updates in later reads.

Availability also has a precise distributed-systems meaning: every request to a non-failed node eventually receives a non-error response. It does not necessarily mean that the response contains the newest value.

Applications must choose semantics according to risk. A social-media reaction counter may tolerate temporary staleness, while an account debit or access-control decision generally requires stronger coordination and auditing.

7. CAP Theorem

The CAP theorem concerns a distributed data store operating during a network partition. It states that a system cannot simultaneously guarantee:

• Consistency: operations behave as if there were one current copy;
• Availability: every request receives a non-error response;
• Partition tolerance: the system continues operating despite communication partition.

Because network partitions are possible in practical distributed systems, the engineering choice during a partition is usually between stronger consistency and higher availability. CAP does not say that a system must permanently choose only two properties, nor does it claim that latency is irrelevant. It applies to particular operations and guarantees under partition conditions.

A consistency-oriented system may reject or delay writes until it can coordinate with a quorum. An availability-oriented system may accept writes on both sides of a partition and reconcile conflicts later. The latter requires conflict-resolution rules and may expose temporary divergent results.

8. PACELC: The Cost Beyond CAP

PACELC extends the analysis:

• P: if there is a partition, choose between Availability and Consistency;
• E: else, when the system is operating normally, choose between Latency and Consistency.

This model highlights that trade-offs exist even without failure. Strong synchronous replication may provide consistency but add cross-zone latency. A system using local replicas may reduce latency while serving stale data.

PACELC is a design lens rather than a universal classification. A product may expose multiple modes: strongly consistent writes, session-consistent reads, and eventually consistent analytical views. The correct choice depends on business consequences, geographic placement, and workload requirements.

9. Consensus and Coordination

Consensus allows distributed participants to agree on one value or one ordered sequence despite failures. A consensus protocol is commonly described using:

• Agreement: correct participants do not decide different values.
• Validity: the decided value is related to a proposed value.
• Termination: correct participants eventually decide, subject to the protocol’s failure assumptions.

Consensus under crash failures is used for leader election, metadata management, configuration changes, and replicated logs. Protocol families such as Raft and Paxos use terms or epochs, quorum voting, and ordered log entries. A leader usually accepts client commands, replicates them to followers, and commits an entry after a sufficient quorum acknowledges it. Followers reject obsolete leaders using term numbers or fencing tokens.

For a cluster of n voting members, a majority quorum is:

For odd n, this is the smallest number greater than half of the members. A five-member cluster therefore requires three votes. It can tolerate two crash failures for continued quorum operation, assuming failures are independent and the remaining members can communicate. Quorum arithmetic alone does not guarantee correctness; membership changes, persistent storage, message ordering, and implementation quality also matter.

Consensus is expensive because it requires communication and durable state. It should coordinate small, critical metadata rather than every high-volume data record when a weaker mechanism is sufficient.

10. Applied Case: A Multi-Region Event Platform

Background

Consider an illustrative online-retail platform collecting clickstream events from users in three geographic regions. The platform supports real-time recommendations, fraud analysis, and regulatory reporting. It receives millions of events per minute, and a regional outage must not stop customer activity.

Problem

A single global partition would create excessive network traffic and a bottleneck. A purely asynchronous design could lose recent events or produce inconsistent fraud decisions. The platform must balance ingestion availability, analytical freshness, and regulatory durability.

Analysis

Events are partitioned by a composite key containing region, event date, and a salted customer identifier. Region and date support locality and retention management; salting reduces hotspots caused by high-volume customers. Each partition is replicated across multiple failure domains. Ingestion acknowledges after a regional quorum confirms durable receipt, while cross-region replication proceeds asynchronously.

The recommendation service reads local, possibly stale aggregates because low latency is important. Fraud decisions use a stronger workflow: decisions are recorded in an ordered log, and duplicate submissions are prevented with idempotency keys. Regulatory data is copied to a controlled archive with retention and access policies.

Approach and outcome

During a regional network partition, local ingestion continues for low-risk events. The system marks data as pending cross-region reconciliation. High-risk transactions are routed through a coordination service or temporarily held. When connectivity returns, events are reconciled using immutable event identifiers and deterministic conflict rules. Monitoring reports replication lag, partition skew, rejected writes, and recovery time.

This design applies PACELC explicitly. During partitions, it favors availability for telemetry but consistency for financial decisions. During normal operation, it accepts some analytical staleness to preserve local latency.

Learning points

1. One consistency policy rarely suits every operation.
2. Composite partition keys must reflect workload and governance requirements.
3. Idempotency is essential when retries can duplicate work.
4. Replication placement must account for correlated failures.
5. Recovery objectives should be measured, not merely promised.

11. Design, Governance, and Operational Considerations

Distributed-system design should begin with explicit service-level objectives. Useful measures include request latency percentiles, throughput, availability, replication lag, recovery point objective, and recovery time objective. Availability over an observation interval can be expressed as:

The formula assumes uptime and downtime are defined consistently and excludes or includes maintenance according to the service agreement. For example, 99 hours of uptime and 1 hour of downtime gives A = 99 ÷ 100 = 0.99, or 99 percent.

Governance is inseparable from architecture. Replicas may create additional copies of personal or regulated data. Designers must address data residency, encryption in transit and at rest, key management, retention, deletion propagation, access control, and auditability. Eventual consistency can complicate deletion guarantees because stale replicas or caches may temporarily retain information.

Operational best practices include:

• use stable, well-distributed partition keys;
• place replicas across independent failure domains;
• make retries safe through idempotency and deduplication;
• use bounded queues and backpressure;
• monitor skew, lag, quorum health, and recovery progress;
• test failover, restore, and network-partition scenarios regularly;
• document consistency guarantees at the API boundary;
• keep consensus groups small and protect their persistent storage;
• separate critical metadata from high-volume analytical data.

Contemporary systems increasingly combine transactional stores, streaming platforms, lakehouses, caches, and AI pipelines. AI workloads intensify locality and scheduling challenges because training data, feature stores, accelerators, and model-serving replicas may span regions. A model trained on inconsistent or incompletely replicated data can produce non-reproducible results. Provenance, versioned datasets, and deterministic replay therefore become part of distributed-data governance.

Challenges and Limitations

No architecture eliminates trade-offs. More replicas increase durability but consume storage and bandwidth. More partitions improve parallelism but increase metadata and coordination overhead. Stronger consistency may increase latency and reduce availability during a partition. Eventual consistency improves elasticity but requires application-level tolerance for stale or conflicting results.

Theoretical models also simplify reality. Amdahl’s Law does not capture changing workload size or network topology. CAP does not specify which consistency model is appropriate. Consensus protocols tolerate only their stated failure model and may fail to make progress when quorum communication is unavailable. Finally, independent-failure assumptions are often violated by power loss, software defects, faulty deployments, or cloud-control-plane failures.

Chapter Summary

Distributed systems divide data and computation across networked components to achieve scale, locality, and resilience. Partitioning controls data placement and parallel work, but poor keys create hotspots and cross-partition coordination. Replication improves durability and availability while introducing consistency and storage costs. Parallel performance is limited by serial work, skew, communication, and coordination overhead.

Fault models clarify which failures a design can tolerate. Consistency describes what clients observe, while availability describes whether requests receive non-error responses. CAP explains the unavoidable consistency–availability choice during network partitions; PACELC adds the latency–consistency trade-off during normal operation. Consensus provides coordinated decisions and ordered logs under defined failure assumptions. Effective designs make these trade-offs explicit, align them with business risk, and validate them through monitoring and failure testing.

Key Terms

Partitioning: Dividing a logical dataset into independently managed subsets.

Sharding: Partitioning data across nodes, commonly for scale.

Replication: Maintaining multiple copies of data.

Consistency: The guarantees governing values and operation ordering observed by clients.

Availability: The property that requests to non-failed components receive responses.

Partition tolerance: Continued operation despite communication separation.

CAP theorem: The result concerning consistency, availability, and partition tolerance during a partition.

PACELC: A framework adding the latency–consistency trade-off when no partition exists.

Quorum: A sufficient subset of participants needed to confirm an operation.

Consensus: Agreement among distributed participants on a value or ordered log.

Data locality: Placing computation near the data it uses.

Hot partition: A partition receiving disproportionate storage or request load.

Partial failure: Failure of some components while others continue operating.

Idempotency: The property that repeating an operation has the same intended effect as performing it once.

Write-ahead log: A durable record written before corresponding state changes are applied.

Checkpoint: A saved recoverable state used to shorten recovery.

Linearizability: A consistency guarantee preserving real-time operation order.

Eventual consistency: Convergence of replicas after updates and communication cease.

Amdahl’s Law: A model relating speedup to serial and parallel portions of a workload.

Failure domain: A set of components likely to fail together.

Review Questions

5 Short Answer Questions

1. What is the purpose of partitioning in a distributed data system?
2. Distinguish synchronous and asynchronous replication.
3. Define a network partition.
4. What does data locality mean?
5. Why is idempotency important when retries are used?

5 Long Answer Questions

1. Compare hash, range, and composite partitioning, including suitable workloads and limitations.
2. Explain CAP and PACELC and show how they influence architectural decisions.
3. Describe a crash-recovery model and the roles of logs, checkpoints, and replay.
4. Explain consensus, quorum, and leader fencing in a replicated-log system.
5. Analyze the relationship between parallelism, Amdahl’s Law, skew, and data movement.

3 Analytical / Application Questions

1. Design a partitioning and replication strategy for a global clickstream platform. Justify the key, replica placement, and consistency policy.
2. A five-member consensus group loses two members and the remaining members cannot communicate with a remote region. Determine whether a majority quorum remains and discuss the consequences for safety and availability.
3. A payment service requires strong correctness, while its recommendation service tolerates stale data. Propose different consistency and failure-handling strategies for the two services and justify them using CAP and PACELC.

Multiple Choice Questions with Answers

Question 1

Which partitioning strategy is generally most suitable for efficient ordered range scans?

A. Hash partitioning

B. Range partitioning

C. Random partitioning

D. Replica partitioning

Correct Answer: B. Range partitioning

Question 2

A system uses ten workers and has a serial fraction of 0.1. According to Amdahl’s Law, the approximate speedup is:

A. 1.11

B. 5.26

C. 9.00

D. 10.00

Correct Answer: B. 5.26

Question 3

What does asynchronous replication primarily increase?

A. Write latency and immediate consistency

B. The need for partition keys

C. The possibility of replication lag

D. The number of serial instructions

Correct Answer: C. The possibility of replication lag

Question 4

During a network partition, CAP describes a fundamental choice between:

A. Durability and compression

B. Availability and consistency

C. Throughput and storage format

D. Encryption and authentication

Correct Answer: B. Availability and consistency

Question 5

For a consensus group of five voting members, a majority quorum contains:

A. One member

B. Two members

C. Three members

D. Five members

Correct Answer: C. Three members

Question 6

Which consistency model requires operations to appear to occur atomically while respecting real-time order?

A. Eventual consistency

B. Causal consistency

C. Sequential consistency

D. Linearizability

Correct Answer: D. Linearizability

Question 7

What is the main purpose of a write-ahead log?

A. To reduce the number of partitions

B. To support recovery by recording intended changes durably

C. To eliminate all network latency

D. To replace access control

Correct Answer: B. To support recovery by recording intended changes durably

Question 8

PACELC adds which trade-off to the CAP discussion?

A. Latency versus consistency when there is no partition

B. Storage versus encryption during a crash

C. Throughput versus partition count during a backup

D. Availability versus compression during normal reads

Correct Answer: A. Latency versus consistency when there is no partition

Question 9

Which condition most directly creates a hot partition?

A. Evenly distributed hash keys

B. A highly popular key receiving disproportionate traffic

C. Replicas placed in separate failure domains

D. A completed checkpoint

Correct Answer: B. A highly popular key receiving disproportionate traffic

Question 10

Which technique makes retrying a request safer when duplicate delivery is possible?

A. Idempotency

B. Randomized partition loss

C. Removing acknowledgements

D. Increasing serial work

Correct Answer: A. Idempotency