Distributed Computing Principles

Introduction

Big data systems process datasets whose volume, velocity, and variety exceed the practical capacity of a single computer. Their solution is distributed computing: the coordinated use of multiple networked computers to store data, execute computation, and provide services. Distribution offers capacity and resilience, but it also introduces communication delay, partial failure, duplicated data, inconsistent views, and difficult coordination.

This chapter develops the principles needed to analyze and design distributed big-data systems. It examines scalability, parallelism, partitioning, replication, fault tolerance, data locality, and consistency. It also compares vertical and horizontal scaling and applies the CAP and PACELC frameworks to practical design decisions.

Learning Objectives

After studying this chapter, you should be able to:

• distinguish centralized, parallel, and distributed computation;
• compare vertical and horizontal scalability;
• explain partitioning, replication, locality, and fault tolerance;
• analyze speedup, throughput, latency, and bottlenecks;
• apply CAP and PACELC to distributed-system choices;
• distinguish strong, causal, read-your-writes, and eventual consistency;
• evaluate a distributed architecture for a big-data application.

1. The Distributed-Computing Model

A distributed system is a collection of independent computers that appears to users as a coordinated system. The computers communicate through a network and may fail independently. A cluster normally contains several roles:

• Clients submit requests or jobs.
• Workers execute computation or serve data.
• Storage nodes hold partitions and replicas.
• Coordinators assign work, maintain metadata, or manage membership.
• Schedulers allocate resources and attempt to balance workloads.

The defining feature is not simply the number of machines. A system is distributed when computation or state is divided across components whose coordination depends on communication. Network communication is slower and less reliable than local memory access. Consequently, designers must treat messages, timeouts, retries, and failures as normal conditions rather than exceptional events.

A distributed system differs from a traditional parallel computer in emphasis. Parallel programs may run on tightly coupled processors with shared memory and highly predictable communication. Distributed systems generally use independent machines, separate memory, and a network. Big-data platforms often combine both: a job is distributed across machines, while each machine performs parallel computation using multiple cores.

Important terminology

• Node: A computer or virtual machine participating in the system.
• Cluster: A group of nodes managed as one computational resource.
• Task: A unit of work assigned to a worker.
• State: Data remembered by a service, such as records, counters, or metadata.
• Coordination: Processes used to agree on membership, ownership, ordering, or state.
• Partial failure: A condition in which some components fail while others continue operating.
• Partition: A logical subset of data assigned to a node or group of nodes.
• Replica: An additional copy of a partition or object.

2. Scalability and Performance

Scalability describes how effectively a system handles increasing workload or resources. Workload may grow through more records, more simultaneous users, larger requests, or stricter response-time requirements.

The principal performance measures are:

• Latency: Time required to complete one operation.
• Throughput: Operations or data units completed per unit of time.
• Capacity: Maximum sustainable workload under stated service requirements.
• Utilization: Fraction of a resource's capacity in use.
• Availability: Proportion of time a service is operational and able to respond acceptably.

A system can have high throughput but poor latency. For example, a batch analytics platform may process millions of records per minute while taking several minutes to return a result. Conversely, an interactive service may have low latency for individual requests but modest total throughput.

Vertical and horizontal scaling

Vertical scaling (scale up) increases the resources of an existing machine: more CPU cores, memory, storage, or network capacity. It is operationally simple and may avoid distributed coordination. However, a single machine has physical limits, upgrades can be expensive, and failure of that machine can remove the entire service.

Horizontal scaling (scale out) adds machines. It can provide large aggregate capacity and often improves fault isolation, but requires partitioning, replication, scheduling, monitoring, and network coordination.

Scaling is not automatically linear. If one node processes a workload in time T₁, and N equivalent nodes process it in time T_N, speedup is:

where S(N) is dimensionless. Ideal linear scaling gives S(N) = N. Parallel efficiency is:

where E(N) ranges from 0 to 1 in the usual interpretation. These formulas assume comparable hardware and a consistently defined workload.

Amdahl's law models a fixed workload when a fraction p can be parallelized and fraction 1-p is serial:

Here N is the number of parallel workers. The serial fraction limits speedup: even as N approaches infinity, speedup approaches 1/(1-p). The model excludes communication overhead unless it is included in the effective serial or parallel portions.

Worked example. Suppose 90% of a job is parallelizable, so p = 0.90, and 10 workers are used. Then:

S(10) = 1 / (0.10 + 0.90/10) = 1/0.19 ≈ 5.26.

Although ten workers are available, the idealized speedup is only about 5.26 because the serial portion and coordination remain significant. Efficiency is approximately 5.26/10 = 0.526, or 52.6%.

Gustafson's law offers a different perspective for scaled workloads: when larger systems process proportionally larger problems, parallel computing can remain useful because the serial fraction need not dominate the total execution time. The two laws are not contradictory; Amdahl addresses a fixed-size problem, while Gustafson emphasizes a scaled problem.

3. Parallelism and Work Decomposition

Parallelism is the simultaneous execution of independent operations. Distributed big-data systems exploit several forms:

1. Data parallelism: The same operation is applied to different data partitions.
2. Task parallelism: Different tasks execute different operations concurrently.
3. Pipeline parallelism: Stages operate concurrently on successive portions of a stream.
4. Query parallelism: Independent parts of a query run on separate workers.

A common batch pattern is map–shuffle–reduce. Map tasks transform input records. A shuffle groups intermediate records by key and transfers them to responsible workers. Reduce tasks aggregate each group. The shuffle often dominates cost because it requires network transfer, serialization, disk spill, and coordination.

Parallelism is effective only when work is sufficiently independent. Dependencies, skew, synchronization, and shared resources reduce gains. If one partition contains most of the records, its worker becomes a straggler, delaying the entire job. Systems address this through better partitioning, adaptive scheduling, speculative execution, or splitting large partitions.

A useful performance model is:

The terms represent computation, network transfer, storage operations, coordination, and waiting or queueing. This is a conceptual decomposition rather than a universal exact equation; some terms overlap in real systems. It is useful because adding processors primarily reduces computation time, not necessarily network, coordination, or storage costs.

4. Partitioning and Data Distribution

Partitioning divides a dataset into subsets that can be stored and processed independently. Good partitioning balances three objectives: even workload, efficient access, and manageable movement when the cluster changes.

Common partitioning strategies

• Hash partitioning: A hash function maps a key to one of N partitions. A simplified rule is partition = hash(key) mod N. It balances many keys but does not preserve ordering.
• Range partitioning: Key intervals are assigned to partitions, such as dates or alphabetical ranges. It supports range queries but may create hotspots when values are unevenly distributed.
• Round-robin partitioning: Records are distributed cyclically. It balances records but may be poor for key-based queries.
• Directory-based partitioning: A metadata service explicitly maps keys or ranges to nodes. It is flexible but introduces metadata-management requirements.
• Consistent hashing: Keys and nodes are placed on a logical ring so that adding or removing a node moves only a subset of keys. Virtual nodes can improve balance.

Partitioning is both a storage and computation decision. A customer-events dataset might be partitioned by customer identifier for customer-specific queries, or by event date for time-window analytics. No partition key is optimal for every query. Designers therefore analyze access patterns, cardinality, skew, update frequency, and data-retention policy.

Data skew occurs when partitions differ greatly in size or workload. It may arise from popular keys, seasonal traffic, or poor range boundaries. Mitigations include salting hot keys, adaptive range splitting, composite keys, pre-aggregation, and workload-aware routing. Salting distributes one hot key across several partitions, but queries must then read and combine those partitions.

5. Replication and Fault Tolerance

Replication stores multiple copies of data or service state. If a partition has replication factor r, then r copies are maintained under the system's replication policy. Replication improves availability and read capacity, but consumes storage and creates consistency obligations.

Replicas may be placed:

• on different processes;
• on different physical machines;
• across racks or availability zones;
• across geographic regions.

Placement across independent failure domains is essential. Three replicas on one physical host do not provide protection against that host's failure.

Replication models include:

• Primary–replica: A primary accepts writes and replicas receive updates. It is simple but can require leader failover.
• Multi-primary: Several nodes accept writes. It improves write locality but requires conflict resolution.
• Leaderless: Clients or coordinators communicate with multiple replicas, often using quorum rules.
• Synchronous replication: A write is acknowledged after specified replicas confirm it. It can provide stronger durability but adds latency.
• Asynchronous replication: Updates are propagated later. It lowers write latency but permits temporary divergence and possible loss of unreplicated data after failure.

A quorum design may use N replicas, write quorum W, and read quorum R. If:

then a read and write quorum must overlap in at least one replica, assuming correct quorum membership and appropriate version handling. This overlap supports visibility of recent writes, but does not by itself guarantee linearizability, conflict-free writes, or protection from all implementation errors. If W + R ≤ N, a read may contact only replicas that have not received the latest write.

Fault tolerance means continuing correct service despite specified failures. Techniques include redundancy, health checks, timeouts, retries, idempotent operations, checkpointing, write-ahead logs, failover, and repair. A retry can accidentally duplicate an operation if the first request succeeded but its response was lost. Therefore, payment and provisioning operations commonly use idempotency keys or transactional deduplication.

A system's failure model must be explicit. Crash failures stop a node; omission failures lose messages; network partitions separate communicating groups; Byzantine failures involve arbitrary or malicious behavior. Most commercial big-data systems primarily model crash and network failures, while security-critical systems may also consider Byzantine behavior.

6. Data Locality

Data locality means executing computation near the data it reads. Moving computation is often cheaper than moving large datasets across a network. A scheduler may therefore prefer a worker that stores a local replica, then a worker in the same rack, then a worker elsewhere.

Locality is not always dominant. If local data are on a slow or overloaded disk, remote computation may be faster. Modern systems use locality-aware scheduling with bounded waiting: they briefly wait for a suitable local resource, then relax locality to avoid delaying the job indefinitely.

Locality also has a logical dimension. Co-locating related partitions can reduce joins, while partitioning both datasets by the same join key can enable a co-partitioned join. Conversely, a poorly chosen partition key can force an expensive all-to-all shuffle.

7. Consistency Models

Consistency specifies what values operations are permitted to observe and how updates become visible. It is not identical to correctness in every application: a social-media “like” counter may tolerate temporary divergence, while an account balance may require strict constraints.

• Strong consistency: A successful read returns a value consistent with a defined global ordering. Linearizability additionally requires that operations appear to occur atomically at points between invocation and response and respect real-time order.
• Sequential consistency: Operations appear in one order that respects each process's program order, but not necessarily real-time order across processes.
• Causal consistency: Causally related operations are observed in causal order; concurrent operations may be observed differently.
• Read-your-writes: A client sees its own completed updates in later reads.
• Monotonic reads: Once a client has seen a value, later reads do not move to an older version.
• Eventual consistency: If updates stop and communication continues, replicas eventually converge to the same value. Eventual consistency alone does not promise a deadline, immediate visibility, or conflict-free semantics.

Eventual consistency is suitable when availability and low latency are more important than immediate agreement, such as product recommendations, presence indicators, and some content feeds. It requires application-level handling of stale reads, conflicts, ordering, and deletion propagation. Version vectors, timestamps, causal metadata, last-write-wins policies, and domain-specific merge functions are common tools. Last-write-wins is simple but can discard a valid concurrent update if clocks or timestamp policies are unsuitable.

8. CAP and PACELC

The CAP theorem concerns a distributed data system subject to a network partition. Its three properties are:

• Consistency (C): Operations behave as if there were one up-to-date copy, often interpreted through linearizable reads and writes.
• Availability (A): Every request to a non-failed node receives a non-error response, without requiring agreement with all other nodes.
• Partition tolerance (P): The system continues operating despite communication loss between parts of the system.

When a partition occurs, a system cannot guarantee both strong consistency and availability for all affected operations. Since real networks can partition, practical designs choose how to behave during partitions: a CP system may reject or delay some requests to preserve consistency, whereas an AP system may accept operations and reconcile divergent states later. CAP does not say that a system must permanently choose only two properties, nor does it mean that latency and performance are irrelevant.

PACELC extends CAP by asking: if there is a Partition, choose between Availability and Consistency; Else, choose between Latency and Consistency. Thus a system can be highly available during normal operation but still pay extra latency for synchronous coordination. PACELC encourages analysis beyond rare partitions.

For example, a globally replicated inventory system may choose consistency over availability when stock is scarce, refusing a purchase if quorum confirmation is unavailable. A recommendation service may choose availability and low latency, returning a slightly stale result during a regional partition. These are business decisions expressed through technical architecture.

9. Case Study: Distributed Retail Analytics and Inventory

The following is an illustrative case. A retailer operates online stores in several regions. It needs near-real-time sales dashboards, personalized recommendations, and accurate inventory reservations.

Background and problem. Sales events arrive continuously from many regions. Analytics queries scan large historical data, while inventory reservations must prevent overselling. A single database creates a capacity bottleneck and exposes the business to a regional outage.

Analysis. Sales events are append-oriented and can be partitioned by event date and region. Dashboards tolerate seconds of delay, making asynchronous replication and eventual convergence acceptable. Inventory reservations have a stronger correctness requirement: two customers must not successfully reserve the last item. Network partitions therefore create a genuine consistency–availability trade-off.

Approach. The architecture uses a horizontally scaled event-ingestion layer. Events are partitioned by region and time, replicated across failure domains, and processed by parallel consumers. A batch and stream analytics layer uses data locality and time-based partitions. Recommendation reads may use local replicas and eventual consistency. Inventory ownership is assigned to a region or partition, and reservation writes use a leader or quorum protocol with idempotency keys. During loss of required coordination, the reservation service rejects or queues requests rather than confirming uncertain stock.

Outcome. Analytics continues with bounded staleness and can recover from worker failure using replicated input and checkpoints. Recommendations remain available during some replication delays. Inventory sacrifices some availability during partitions, protecting the invariant that confirmed reservations do not exceed available stock.

Learning points. One organization may use different consistency models for different data. Partition keys should reflect access patterns. Replication placement must follow failure domains. “Distributed” is not a single architectural choice: each workload requires an explicit decision about latency, availability, durability, and correctness.

10. Design Practices, Governance, and Contemporary Developments

Effective distributed-system design begins with workload characterization: request rates, data growth, read/write ratios, access patterns, acceptable staleness, recovery objectives, and failure assumptions. Designers should define service-level objectives for latency, availability, and durability rather than using vague claims such as “high performance.”

Recommended practices include:

1. partition using measured access patterns and monitor skew;
2. replicate across independent failure domains;
3. make retries safe through idempotency and deduplication;
4. use timeouts and bounded retries to prevent retry storms;
5. monitor lag, queue depth, hot partitions, repair status, and tail latency;
6. test node, rack, zone, and network failures through controlled experiments;
7. document consistency guarantees at the API boundary;
8. plan rebalancing, schema evolution, backup, restoration, and data deletion;
9. protect data with encryption, authentication, authorization, and audit logs;
10. minimize unnecessary copies of personal or regulated data.

Governance is especially important in big-data systems. Replication can move personal data across jurisdictions. Retention and deletion requirements must account for backups, caches, replicas, and derived datasets. Access controls should apply consistently across partitions and replicas. Automated reconciliation or AI-based anomaly detection may help identify skew, failures, or unusual access, but it must be observable and assessed for false positives, privacy impact, and discriminatory effects. AI does not remove the need for explicit consistency and recovery guarantees.

Contemporary systems increasingly combine batch and stream processing, use disaggregated compute and storage, employ cloud-based elastic clusters, and support geo-distributed databases. These developments improve flexibility but make network cost, data sovereignty, observability, and egress pricing more important. The fundamental principles remain unchanged: divide work carefully, place data deliberately, replicate according to failure goals, and select consistency according to business correctness.

Chapter Summary

Distributed computing coordinates independent machines to process data and provide services despite communication delay and partial failure. Vertical scaling upgrades one machine; horizontal scaling adds machines and demands partitioning, scheduling, replication, and coordination. Parallelism improves performance only when serial work, skew, network transfer, storage, and synchronization are controlled. Amdahl's law demonstrates why a serial fraction limits speedup.

Partitioning determines where data and computation occur. Hash, range, round-robin, directory-based, and consistent-hashing strategies suit different access patterns. Replication increases availability and resilience but introduces update propagation and conflict-management costs. Fault tolerance depends on redundancy, failure detection, safe retries, checkpoints, repair, and explicit failure models. Data locality reduces network movement and can make joins more efficient.

Consistency ranges from strong and linearizable behavior to eventual convergence. CAP explains the trade-off between consistency and availability during a network partition, while PACELC adds the normal-case latency–consistency trade-off. A well-designed system applies different choices to different workloads, protects business invariants, and documents its guarantees.

Key Terms

Scalability: Ability to handle increased workload or resources effectively.

Vertical scaling: Increasing the capacity of an existing machine.

Horizontal scaling: Adding machines to increase aggregate capacity.

Latency: Time required to complete an operation.

Throughput: Amount of work completed per unit of time.

Parallelism: Simultaneous execution of independent work.

Partitioning: Dividing data into independently managed subsets.

Data skew: Uneven distribution of data or workload among partitions.

Replication: Maintaining multiple copies of data or state.

Replication factor: Number of maintained copies of a partition or object.

Fault tolerance: Ability to continue correct operation despite specified failures.

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

Data locality: Executing computation close to the data it uses.

Consistency: Rules describing which values and operation orders clients may observe.

Eventual consistency: Guarantee that replicas converge if updates stop and communication continues.

CAP theorem: Framework describing the consistency–availability trade-off during partitions.

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

Quorum: Required subset of replicas for an operation to proceed.

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

Data skew: Unequal partition sizes or workloads that create bottlenecks.

Review Questions

Short Answer Questions

1. What is the difference between vertical and horizontal scaling?
2. Why can a distributed system experience partial failure?
3. Define data locality and explain why it matters.
4. What does eventual consistency guarantee, and what does it not guarantee?
5. What is the purpose of a replication factor?

Long Answer Questions

1. Explain how partitioning, replication, and fault tolerance work together in a big-data platform.
2. Compare hash, range, and consistent-hash partitioning, including a suitable use case for each.
3. Explain Amdahl's law and discuss why adding workers may fail to produce linear speedup.
4. Compare strong, causal, and eventual consistency in terms of visibility, latency, and application suitability.
5. Explain CAP and PACELC and distinguish their treatment of partition-time and normal-operation trade-offs.

Analytical / Application Questions

1. A batch job has 80% parallelizable work and uses 8 workers. Use Amdahl's law to calculate its idealized speedup and efficiency. Explain the limitations of your result.
2. Design a partitioning and replication strategy for a global event-streaming service. Justify the key, replica placement, consistency model, and response to a regional partition.
3. An inventory service must never confirm more reservations than available stock, but customers also expect regional availability. Analyze the CAP and PACELC choices and propose a defensible architecture.

Multiple Choice Questions with Answers

Question 1

Which action is an example of horizontal scaling?

A. Adding memory to an existing server

B. Replacing a CPU with a faster model

C. Adding additional worker nodes to a cluster

D. Increasing the size of one disk

Correct Answer: C. Adding additional worker nodes to a cluster

Question 2

If 90% of a workload is parallelizable, what is the Amdahl speedup with 10 workers?

A. Approximately 2.90

B. Approximately 5.26

C. Exactly 9.00

D. Exactly 10.00

Correct Answer: B. Approximately 5.26

Question 3

Which partitioning strategy naturally supports ordered range queries?

A. Range partitioning

B. Round-robin partitioning

C. Random replication

D. Pure task partitioning

Correct Answer: A. Range partitioning

Question 4

What is a principal risk of data skew?

A. All replicas become strongly consistent

B. One worker becomes a bottleneck or straggler

C. Network communication becomes unnecessary

D. Partition keys are automatically encrypted

Correct Answer: B. One worker becomes a bottleneck or straggler

Question 5

What is the main purpose of replication?

A. To eliminate all coordination

B. To create additional copies for resilience or capacity

C. To guarantee zero network latency

D. To reduce the logical size of a dataset

Correct Answer: B. To create additional copies for resilience or capacity

Question 6

Under the usual quorum condition, which relationship promotes overlap between a read quorum and a write quorum?

A. W + R < N

B. W − R = N

C. W + R > N

D. W × R = 0

Correct Answer: C. W + R > N

Question 7

What does eventual consistency guarantee?

A. Every read immediately returns the newest global value

B. Replicas converge if updates stop and communication continues

C. All operations are totally ordered in real time

D. No conflicts can ever occur

Correct Answer: B. Replicas converge if updates stop and communication continues

Question 8

During a network partition, CAP describes a trade-off primarily between which properties?

A. Latency and throughput

B. Partition tolerance and encryption

C. Consistency and availability

D. Storage and computation

Correct Answer: C. Consistency and availability

Question 9

What additional trade-off does PACELC emphasize when there is no partition?

A. Latency versus consistency

B. Replication versus authentication

C. Throughput versus storage format

D. Partitioning versus serialization

Correct Answer: A. Latency versus consistency

Question 10

Which practice makes retrying a potentially duplicated request safer?

A. Removing all timeouts

B. Using an idempotency key

C. Increasing partition skew

D. Disabling failure detection

Correct Answer: B. Using an idempotency key