2. Distributed Computing Principles

Introduction and Context

Big data platforms are built on a simple but demanding premise: data and computation can be distributed across many networked computers rather than concentrated on one powerful machine. This approach makes it possible to store petabytes of information, process millions of events per second, and continue operating despite hardware failures. However, distributing a workload introduces problems that do not arise, or arise less severely, in a single-machine system. Machines fail independently, networks introduce delay, clocks disagree, data may be duplicated, and no component has a complete and instantly current view of the entire system.

Distributed computing principles provide the conceptual foundation for technologies such as Hadoop, Apache Spark, Apache Kafka, distributed databases, object stores, cloud data warehouses, and stream-processing platforms. A data scientist who understands only the programming interface of these systems may produce applications that are correct on small datasets but unreliable, slow, or expensive at scale. Understanding partitioning, replication, parallelism, fault tolerance, locality, consistency, availability, and distributed processing models is therefore essential for both technical implementation and architectural decision-making.

This chapter examines how distributed systems divide work and data, how they respond to failure, and why design choices involve trade-offs rather than universally superior solutions. It also connects theoretical models such as CAP and PACELC with practical choices in batch analytics, streaming, transactions, and cloud-scale data engineering.

Learning Objectives

After studying this chapter, students should be able to:

1. Explain the purpose and operation of partitioning, replication, parallelism, and data locality.

2. Analyse how distributed systems detect, tolerate, and recover from failures.

3. Distinguish strong, eventual, causal, and other consistency models.

4. Apply CAP and PACELC to distributed data-system design decisions.

5. Compare batch, stream, message-driven, and actor-oriented processing models.

6. Evaluate distributed architectures according to performance, cost, reliability, governance, and business requirements.

7. Identify common implementation mistakes and recommend suitable best practices.

From One Computer to a Distributed System

A distributed system is a collection of independent computers that coordinates its actions through a network and presents some degree of unified behaviour to users or applications. The computers may be located in one data centre, across several availability zones, or across continents. The objective is usually one or more of the following: increased capacity, improved performance, higher availability, geographic reach, or reduced cost through horizontal scaling.

Vertical scaling increases the resources of one machine by adding processors, memory, or storage. Horizontal scaling adds machines. Horizontal scaling is attractive because commodity or cloud instances can be added incrementally, but it creates coordination overhead. A request may need information held on several machines; results must be combined; and failures must be expected rather than treated as exceptional events.

Two properties distinguish distributed systems from merely parallel programs. First, communication occurs through a network, which is slower and less reliable than communication within a computer. Second, components have partial knowledge. A node cannot instantly know whether another node is processing slowly, disconnected, or permanently failed. This uncertainty explains why timeout values, retries, leases, heartbeats, consensus protocols, and failure detectors are central to distributed-system design.

Dividing Data and Work

Partitioning and Sharding

Partitioning divides a dataset into subsets called partitions. Each partition is managed or processed by a particular node or group of nodes. In database terminology, horizontal partitioning is often called sharding. Partitioning enables storage and computation to grow across machines, but it also determines how efficiently queries and jobs can be executed.

A range partition assigns records according to an ordered key, such as date or customer identifier. Range partitioning supports efficient queries over intervals, but poorly chosen ranges can create hotspots. For example, if most incoming events have the current date, a partition based on date may receive nearly all writes.

Hash partitioning applies a hash function to a key and distributes records among buckets. It generally spreads load more evenly, although it does not naturally support range queries. Consistent hashing is useful in systems where nodes are added or removed because it can limit the amount of data that must move during rebalancing.

Round-robin partitioning assigns successive records to successive nodes. It can balance volume effectively when records have similar processing costs, but it may be unsuitable when related records should be processed together. A composite or co-partitioning strategy can place records with the same key in corresponding partitions, reducing the cost of joins.

Partitioning introduces several important concerns:

• Partition key selection: The key should distribute data and workload evenly while supporting common access patterns.
• Skew: A small number of keys may account for a disproportionate amount of data or activity.
• Rebalancing: Adding or removing nodes may require data movement.
• Ordering: Global ordering is expensive; many systems provide ordering only within a partition.
• Cross-partition operations: Joins, aggregations, and transactions involving multiple partitions require coordination.

Partitioning is therefore not simply a storage decision. It is a data-modelling and workload-design decision.

Replication

Replication maintains multiple copies of data or service state. A replica may be placed on another machine, rack, availability zone, or geographic region. Replication improves availability and can increase read capacity, but it consumes storage and introduces the problem of keeping copies sufficiently aligned.

In synchronous replication, a write is acknowledged only after specified replicas have confirmed it. This offers stronger durability and consistency but increases latency and may reduce availability when replicas cannot communicate. In asynchronous replication, the primary acknowledges the write before all replicas receive it. This improves responsiveness but creates a window in which a failure may cause recent data to be unavailable or lost.

Replication factor describes the number of copies maintained. A higher replication factor can tolerate more failures, but it increases storage cost, network traffic, repair time, and the complexity of maintaining consistency. Replication is not equivalent to backup: replicas may reproduce accidental deletion, corruption, or ransomware. Independent backups and recovery testing remain necessary.

Common replication arrangements include primary–secondary replication, multi-primary replication, and quorum-based replication. In primary–secondary systems, one node usually accepts writes while secondary nodes serve reads or take over after failure. Multi-primary systems permit writes at several locations but must resolve conflicts. Quorum systems require a sufficient number of replicas to participate in reads and writes. If a system has N replicas, write quorum W, and read quorum R, the condition R + W > N can ensure overlap between a read and a completed write under suitable assumptions. The exact guarantees depend on timing, conflict handling, and implementation details.

Parallelism

Parallelism is the simultaneous execution of multiple operations. Data parallelism applies the same operation to different partitions, while task parallelism executes different operations concurrently. A distributed analytics job may use both: one set of workers scans partitions while another stage aggregates intermediate results.

The theoretical speedup from adding processors is limited by the proportion of work that cannot be parallelised. Amdahl’s Law expresses this limitation. If a fraction s of a job is serial and p processors execute the parallel portion, the maximum speedup is approximately:

\[ S(p) = \frac{1}{s + \frac{1-s}{p}} \]

For large-scale data processing, communication and coordination add further overhead. Consequently, doubling the number of nodes does not necessarily halve execution time. Small tasks may be dominated by scheduling overhead, while excessive parallelism may create too many files, network transfers, or metadata operations.

A well-designed distributed job seeks an appropriate granularity: partitions should be large enough to amortise overhead but small enough to permit load balancing and failure recovery. Stragglers—workers that progress substantially more slowly than others—can delay an entire stage. Causes include hardware contention, skewed partitions, garbage collection, noisy neighbours, or network problems.

Data Locality

Data locality means executing computation near the data it needs. If a worker can read data from local storage, it avoids network transfer. Rack locality and zone locality are weaker forms of locality but may still reduce cost and latency. Distributed batch frameworks commonly schedule tasks preferentially where input blocks are stored, while cloud object storage changes the calculation because storage and compute are often separated.

Locality is valuable, but it should not be treated as an absolute rule. A local worker with insufficient capacity may be slower than a remote worker with available resources. Cloud systems may also prioritise elastic compute, independent scaling, and operational simplicity over physical locality. The practical objective is to minimise total cost and completion time, considering network bandwidth, storage architecture, data size, and workload behaviour.

Failure as a Normal Operating Condition

A distributed system must assume that machines, disks, processes, networks, and software components can fail independently. Failures may be crash failures, in which a component stops responding; omission failures, in which messages are lost; timing failures, in which responses arrive too late; or Byzantine failures, in which a component behaves arbitrarily or maliciously. Most big data platforms primarily design for crash and network failures, while security-sensitive systems may also consider Byzantine behaviour.

Fault tolerance is the ability to continue providing an acceptable service despite failures. Techniques include replication, checkpointing, retry, task re-execution, write-ahead logging, quorum protocols, failover, and graceful degradation.

Retries are useful for transient failures but dangerous when operations are not idempotent. Repeating a payment or increment operation may produce an incorrect result unless the operation has a unique request identifier or transactional protection. Exponential backoff and jitter prevent many clients from retrying simultaneously and creating a retry storm.

Checkpointing records intermediate state so that a failed computation can restart from a recent point rather than from the beginning. In batch processing, a framework may recompute a lost partition from an immutable source. In stream processing, checkpoints may record offsets and operator state. The guarantee achieved depends on how input offsets, state updates, and outputs are coordinated.

Exactly-once processing is often used imprecisely. It may mean that a framework executes a computation once logically, even if tasks are retried internally. Achieving exactly-once effects across an external database or API requires transactions, idempotent writes, or a carefully designed transactional sink. At-least-once delivery combined with idempotent processing is frequently more practical than attempting universal exactly-once behaviour.

Consistency, Availability, and the CAP Perspective

Consistency describes what values users can observe when data is replicated or updated concurrently. Strong consistency, often expressed as linearizability, gives the impression that each operation occurs atomically at one point between its invocation and response. A read returns the latest completed write according to a single system-wide order. This is intuitive but may require coordination and can increase latency.

Eventual consistency guarantees that, if updates stop, replicas will eventually converge. A read may temporarily return an older value. This model is often suitable for social feeds, cached content, and some recommendation systems, but it is inappropriate where stale information could cause unacceptable harm or financial error.

Causal consistency preserves cause-and-effect relationships. If one event could have influenced another, all observers see them in that order, although unrelated events may be seen in different orders. Session guarantees, such as read-your-writes and monotonic reads, provide useful user-level assurances without requiring full linearizability.

Availability can have different meanings. In CAP, availability means that every request received by a non-failing node eventually receives a non-error response. In operational discussions, availability may instead refer to uptime, which also depends on maintenance, performance thresholds, and service-level objectives. These meanings should not be confused.

The CAP theorem states that a distributed data system cannot simultaneously guarantee consistency, availability, and partition tolerance in the presence of a network partition. A partition is a condition in which groups of nodes cannot communicate reliably. Since network partitions must generally be considered possible, a system must choose whether to reject or delay some operations to preserve consistency, or continue responding with potentially divergent results to preserve availability.

CAP is not a classification that permanently labels a system as simply CP or AP. Different operations, configurations, and failure periods may use different policies. Moreover, partition tolerance is not an optional preference in a geographically distributed system. The practical choice is usually between consistency and availability during a partition, with the decision driven by business consequences.

PACELC: The Cost of Normal Operation

PACELC extends CAP by asking what happens when the system is operating normally. Its formulation is: if there is a Partition, choose between Availability and Consistency; Else, choose between Latency and Consistency. Even without a partition, synchronous coordination across replicas may increase latency. A system that prefers low latency may accept stale reads or asynchronous replication, while a consistency-oriented system may incur additional communication before acknowledging an operation.

PACELC is particularly useful for cloud and global applications. A globally distributed catalogue may prefer low-latency local reads and tolerate brief staleness. A financial ledger may require stronger consistency despite increased latency. Neither choice is universally correct; the architecture should reflect the cost of incorrect, delayed, or unavailable results.

Distributed Processing Models

Batch Processing

Batch processing collects bounded input and processes it as a finite job. MapReduce provides a classic model: map functions transform records into intermediate key–value pairs, and reduce functions aggregate values with the same key. A shuffle moves intermediate data across the network so that related keys reach the same reducers. The shuffle is often the most expensive stage because it requires network transfer, sorting, disk I/O, and coordination.

Modern data engines generalise this idea through directed acyclic graphs of transformations. They may optimise execution by combining operations, avoiding unnecessary materialisation, and recovering only lost partitions. Batch processing is suitable for historical reporting, large-scale feature generation, data-lake transformation, and periodic model training. Its limitations include higher latency and difficulty responding immediately to individual events.

Stream Processing

Stream processing handles unbounded, continuously arriving data. Events are processed as they arrive, often using windows. Tumbling windows are fixed and non-overlapping; sliding windows overlap; session windows group events separated by periods of inactivity. Event time refers to when an event occurred, whereas processing time refers to when the system handled it. Because events can arrive late or out of order, robust systems use watermarks to estimate when sufficiently complete event-time data has been observed.

Stream systems must define delivery and state semantics. They may provide at-most-once, at-least-once, or exactly-once processing under specified conditions. Stateful operators require durable state, checkpointing, and recovery procedures. Stream processing is useful for fraud detection, monitoring, personalisation, industrial telemetry, and real-time operational decisions.

Message-Driven and Actor-Oriented Processing

In message-driven systems, components communicate through queues or logs rather than direct synchronous calls. This decouples producers and consumers, absorbs bursts, and supports replay when messages are retained. Partitioned logs can preserve ordering within a partition while allowing parallel consumption across partitions.

Actor-oriented systems model computation as independent actors with private state that communicate by messages. Actors can be distributed and supervised, making the model useful for highly concurrent services. However, message ordering, duplicate messages, backpressure, and state recovery still require explicit design.

Practical Design Scenario: A Global Retail Analytics Platform

The following is an illustrative case study.

Background

A retail organisation operates online stores in several regions. It receives clickstream events, orders, inventory updates, and delivery information. Business users require daily sales reports, while fraud analysts need near-real-time alerts. Customers expect product pages to load quickly in every region.

Situation and Problem

An initial centralised database becomes a bottleneck during promotional events. Reporting queries interfere with order processing, and a regional network outage prevents some users from accessing the service. The organisation must support high write volume, tolerate failures, provide timely analytics, and avoid excessive infrastructure cost.

Analysis

The workloads have different requirements. Orders and payments need strong correctness and auditable state. Product recommendations can tolerate some staleness. Clickstream data is high volume and can be processed asynchronously. Daily reports can use batch computation over historical data, whereas fraud detection requires streams.

A single consistency policy would be inefficient. Strong coordination for every clickstream event would increase cost and latency without producing corresponding business value. Conversely, eventual consistency for payment state would create unacceptable risk.

Approach and Strategy

The organisation designs separate but integrated paths. Orders are partitioned by customer or order identifier and stored with replicated transactional data. A durable event log receives order and clickstream events. Stream processors consume the log for fraud signals and operational monitoring. A data lake stores immutable raw events, while batch jobs produce curated analytical tables and aggregates.

Product data is replicated regionally for low-latency reads. Updates use a defined conflict policy and may be temporarily stale. Payment confirmation remains governed by a strongly consistent transactional service. Partition keys are tested against promotional workloads to identify hot keys. Consumers use idempotent event processing, and checkpoints allow recovery after worker failure. Backups are maintained independently of operational replicas.

The architecture also separates resource pools: interactive transactions, stream processing, and batch analytics do not compete blindly for the same capacity. Monitoring measures lag, partition skew, failed tasks, retry rates, checkpoint age, replica divergence, and end-to-end freshness.

Outcome

The design allows reporting workloads to scale independently from customer transactions. Stream processing provides timely fraud indicators, while batch processing produces reconciled reports. During a regional outage, traffic can be redirected according to service-specific policies: some read services continue with replicated data, while sensitive operations fail safely rather than accepting uncertain state.

Learning Points

The case illustrates that distributed architecture is a portfolio of trade-offs rather than a single technology choice. Partitioning must reflect access patterns; replication must reflect failure and consistency requirements; processing models should match latency needs; and business risk must determine where availability or consistency receives priority.

Challenges, Risks, and Common Mistakes

A frequent mistake is choosing a partition key based only on logical meaning rather than workload distribution. A natural key may create severe skew. Another is assuming that replication eliminates data loss; asynchronous replicas may lag, and replicas may reproduce corruption. Engineers also underestimate cross-partition operations, which can turn an apparently simple query into a costly distributed join.

Unbounded retries can overload a recovering service. Non-idempotent consumers may duplicate business effects. Ignoring late events produces inaccurate stream results. Treating processing time as event time can misclassify transactions that arrive late. Excessive small files create metadata and scheduling overhead in data lakes, while excessively large partitions reduce parallelism and make recovery expensive.

Distributed systems also create observability challenges. Average latency can conceal tail latency, where a small percentage of requests take much longer. Monitoring should include percentiles, queue depth, consumer lag, error rates, replica health, data freshness, and saturation. Testing should include node loss, network delay, partition, clock skew, duplicate delivery, schema change, and recovery from backup—not merely normal operation.

Governance, Security, and Professional Responsibility

Distributed data may cross organisational or national boundaries. Partition and replication policies must therefore consider data residency, privacy, retention, encryption, access control, and legal discovery. Replicating personal data into additional regions increases the number of locations that must be secured and governed.

Consistency choices can also have ethical consequences. A stale fraud or eligibility decision may unfairly affect individuals. Analytics teams should document freshness assumptions, failure behaviour, reconciliation processes, and the consequences of incorrect results. Audit trails should distinguish original events, corrected events, and derived outputs. Professionals should not describe a system as “real time,” “exactly once,” or “highly available” without defining the measurement and operating conditions.

Emerging Directions

Cloud-native platforms increasingly separate storage and compute, allowing each to scale independently. This reduces the importance of physical locality in some workloads but increases dependence on network performance and data-transfer cost. Lakehouse architectures combine large-scale object storage with transactional metadata and analytical table formats. Serverless processing hides much infrastructure management but does not remove partitioning, consistency, or cost considerations.

Modern stream processors increasingly support unified batch and stream abstractions, incremental materialised views, and event-time correctness. Geographically distributed databases offer configurable consistency and placement policies. Hardware acceleration, disaggregated memory, and specialised data-processing engines may improve performance, but they also make resource scheduling and failure recovery more complex.

Artificial intelligence workloads add new distributed requirements. Training may use data parallelism, model parallelism, or pipeline parallelism. Parameter synchronisation, checkpoint size, network bandwidth, and reproducibility become central concerns. At the same time, automated tuning systems can adjust partition counts, cluster capacity, and query plans; such automation should be constrained by observability, cost controls, and governance rules.

Best-Practice Principles

A sound distributed design normally follows these principles:

1. Begin with workload requirements: volume, velocity, latency, correctness, availability, retention, and geographic scope.

2. Choose partition keys from observed access and write patterns, not intuition alone.

3. Design explicitly for skew, rebalancing, stragglers, duplicate messages, and partial failure.

4. Match consistency to business risk and document the guarantees users actually receive.

5. Prefer idempotent operations and durable, independently tested recovery mechanisms.

6. Separate workloads with different performance and reliability objectives.

7. Measure end-to-end freshness and tail latency, not only throughput and averages.

8. Treat replicas, backups, and audit records as different controls with different purposes.

9. Test failure and recovery continuously in realistic environments.

10. Revisit architecture as data volume, query patterns, regulations, and business consequences change.

Chapter Summary

Distributed computing enables big data platforms to scale storage and computation across many machines, but it introduces network delay, partial knowledge, coordination costs, and independent failures. Partitioning divides data for scalable access and processing; replication improves resilience and availability but creates consistency and storage costs. Parallelism reduces execution time only when work is sufficiently divisible and coordination overhead is controlled. Data locality can reduce network movement, although cloud separation of storage and compute changes its role.

Fault tolerance depends on replication, retries, checkpointing, re-execution, idempotence, and recovery design. Consistency ranges from strong guarantees to eventual convergence, with causal and session guarantees occupying useful intermediate positions. CAP explains the consistency–availability choice during network partitions, while PACELC highlights the latency–consistency trade-off during normal operation. Batch, stream, message-driven, and actor-oriented models serve different workload requirements. Effective architecture therefore requires explicit trade-offs based on business risk, performance, cost, governance, and operational reality.

Key Terms

Partitioning: Dividing data or work into subsets handled by different nodes.

Sharding: Horizontal partitioning of a dataset across independent storage nodes.

Replication: Maintaining multiple copies of data or state.

Replication factor: The number of copies maintained for a data item or partition.

Data locality: Processing data near the location where it is stored.

Data skew: Uneven distribution of records or workload across partitions.

Parallelism: Simultaneous execution of multiple computations.

Straggler: A slow worker that delays completion of a distributed operation.

Fault tolerance: The ability to continue operating despite component failures.

Idempotence: A property whereby repeating an operation produces the same effective result as performing it once.

Checkpoint: Persisted execution or state information used for recovery.

Consistency: The rules governing values and ordering visible to users of replicated data.

Eventual consistency: A guarantee that replicas converge when updates cease.

Linearizability: A strong consistency model in which operations appear to occur atomically in real-time order.

Availability: The ability of a system to respond successfully according to a defined service expectation.

Network partition: A communication failure separating components or groups of components.

CAP theorem: The principle that consistency, availability, and partition tolerance cannot all be guaranteed simultaneously during a partition.

PACELC: A model describing the consistency–availability choice during partitions and the latency–consistency choice otherwise.

Watermark: A stream-processing indicator of the progress of event-time completeness.

Quorum: A sufficient subset of replicas required to approve or serve an operation.

Review Questions

Short Answer Questions

1. Why does horizontal scaling create coordination challenges?

2. Distinguish range partitioning from hash partitioning.

3. What is the difference between replication and backup?

4. Why can retries be unsafe for non-idempotent operations?

5. What role does a watermark play in stream processing?

Long Answer Questions

1. Explain how partition-key selection affects performance, scalability, and data skew.

2. Discuss synchronous and asynchronous replication, including their effects on latency, durability, and availability.

3. Explain CAP and PACELC and show how they guide distributed-system design.

4. Compare batch processing and stream processing with respect to input, latency, state, failure recovery, and applications.

5. Analyse the role of fault tolerance mechanisms in a distributed analytics platform.

Analytical / Application Questions

1. A system receives most writes for the current date and uses date-range partitioning. Diagnose the likely problem and propose alternatives.

2. A global recommendation service experiences high latency when it waits for all replicas before serving reads. Using PACELC, recommend a revised policy and identify its risks.

3. Design a high-level architecture for a fraud-detection platform that combines transactional correctness, real-time alerts, historical reporting, replication, and failure recovery.

22. Multiple Choice Questions with Answers

Question 1

A data platform uses hash partitioning on customer_id, but one customer generates most of the events. What problem is most likely to occur?

A. Improved global ordering

B. Partition skew and a hot partition

C. Automatic strong consistency

D. Elimination of replication overhead

Correct Answer: B. Partition skew and a hot partition

Question 2

Which statement best describes asynchronous replication?

A. A write is acknowledged only after every replica confirms it

B. Replicas can never diverge

C. The primary may acknowledge a write before all replicas receive it

D. It removes the need for backups

Correct Answer: C. The primary may acknowledge a write before all replicas receive it

Question 3

Why is data locality often beneficial in distributed batch processing?

A. It eliminates the need for partitioning

B. It reduces network transfer of input data

C. It guarantees strong consistency

D. It prevents all worker failures

Correct Answer: B. It reduces network transfer of input data

Question 4

A distributed job has a large serial component. Which principle explains why adding many workers produces limited speedup?

A. CAP theorem

B. Amdahl’s Law

C. Quorum consensus

D. Eventual consistency

Correct Answer: B. Amdahl’s Law

Question 5

During a network partition, a system rejects writes rather than risk conflicting values. Which trade-off is it prioritising?

A. Availability over consistency

B. Consistency over availability

C. Latency over durability

D. Parallelism over locality

Correct Answer: B. Consistency over availability

Question 6

According to PACELC, what trade-off exists when there is no network partition?

A. Latency versus consistency

B. Availability versus partition tolerance

C. Replication versus backup

D. Batch versus streaming

Correct Answer: A. Latency versus consistency

Question 7

Which design most directly supports safe reprocessing of duplicated messages?

A. Non-idempotent updates without identifiers

B. Idempotent operations with unique event identifiers

C. Disabling checkpoints

D. Using only local storage

Correct Answer: B. Idempotent operations with unique event identifiers

Question 8

In stream processing, event time differs from processing time because event time records:

A. When the system finished computation

B. When the event actually occurred

C. When a replica was created

D. When a partition was rebalanced

Correct Answer: B. When the event actually occurred

Question 9

What is the primary purpose of a checkpoint in a stateful distributed computation?

A. To increase data skew

B. To record recovery information and avoid restarting from the beginning

C. To guarantee that every external side effect is exactly once

D. To replace independent backups

Correct Answer: B. To record recovery information and avoid restarting from the beginning

Question 10

A retail platform uses strong consistency for payments but allows stale product recommendations. What principle does this illustrate?

A. Every workload should use identical consistency guarantees

B. Consistency should match business risk and workload requirements

C. Replication is unnecessary for analytical data

D. Batch processing is always superior to streaming

Correct Answer: B. Consistency should match business risk and workload requirements