Managing

Transactions And
Data Integrity

MODULE 5
 Syllabus:

• Managing Transactions And Data Integrity:

RDBMS and ACID Isolation Levels and Isolation Strategies Distributed ACID Systems Consistency
Availability Partition Tolerance Upholding CAP Compromising on Availability Compromising on Partition
Tolerance Compromising on Consistency Implementations in a Few NoSQL Products Distributed
Consistency in MongoDB Eventual Consistency in CouchDB Eventual Consistency in Apache
Cassandra Consistency in Membase Summary
• PERFORMANCE TUNING:
Goals of Parallel Algorithms, The Implications of Reducing Latency, How to Increase Throughput,
Linear Scalability, Influencing Equations, Amdahl’s Law, Little’s Law, Message Cost Model, Partitioning,
Scheduling in Heterogeneous Environments, Additional Map-Reduce Tuning, Communication
Overheads, Compression, File Block Size, Parallel Copying.
• SLT: HBase Coprocessors, Leveraging Bloom Filters.
• Textbook1: Chapter9, Chapter16
 Property Description

A transaction is treated as a single,

RDBMS and ACID Atomicity
indivisible unit. It either fully completes
(commits) or fully fails (aborts/rolls
back). There are no partial results.
A transaction can only take the database
• The ACID properties form the core from one valid state to another. It must
Consistency
foundation for transaction reliability in adhere to all defined rules, constraints (like
traditional RDBMS, ensuring data integrity unique keys, foreign keys), and triggers.
even with concurrent operations or system Concurrent transactions must execute in
failures.2 such a way that they appear to be
executed serially (one after the other).
Isolation
This prevents one transaction from being
affected by others running at the same
time.
Once a transaction is committed, its
changes are permanent and survive
subsequent system failures (like crashes or
Durability
power outages), typically ensured by
writing to persistent storage like a disk or
RDBMS and ACID
• As an example, consider two processes, X and Y,
modifying the value of a field V, which holds an initial
value V0. Say X reads the value V0 and wants to update
the value to V1 but before it completes the update Y
reads the value V0 and updates it to V2. Now when X
wants to write the value V1 it finds that the original value
has been updated. In an uncontrolled situation, X would
overwrite the new value that Y has written, which may
not be desirable. Look at Figure 9-1 to view the stated
use case pictorially. Isolation assures that such
discrepancies are avoided. The different levels and
strategies of isolation are explained later in a following
section
 Isolation Level
Isolation Levels and (from weakest to
strongest)
Anomalies
Prevented
Anomalies
Possible
Description

Isolation Strategies : Dirty Read, Non-

A transaction can
Read read uncommitted
None Repeatable Read,
Uncommitted data from another
• To balance data integrity with performance and Phantom Read
transaction.
concurrency, the SQL standard defines four transaction
A transaction can
Isolation Levels. A lower isolation level allows for only read data that
better concurrency but risks specific "read phenomena" Non-Repeatable
has been
or anomalies. Read Committed Dirty Read Read, Phantom
committed. Most
Read
Key Read Phenomena (Anomalies): common default
level.
• Dirty Read: A transaction reads data written by a
concurrently running transaction that is not yet A transaction can
committed. reread data and find
the same values, as
• Non-Repeatable Read (Fuzzy Read): A transaction Dirty Read, Non- if it took a snapshot
reads a row twice and gets a different value because Repeatable Read Phantom Read
Repeatable Read of the data rows.
another transaction modified and committed the New rows
change between the two reads. (phantoms) can still
• Phantom Read: A transaction re-executes a query appear.
that returns a set of rows and gets a different set of Highest level.
rows because another committed transaction inserted Transactions
or deleted rows matching the query's criteria. execute completely
Dirty Read, Non- isolated, as if they
Serializable Repeatable Read, None were running one
Phantom Read after the other.
Guarantees true
serial execution
DISTRIBUTED ACID SYSTEMS:

The core challenge in Distributed ACID Systems is reconciling the requirement for strong
consistency (guaranteed by ACID) with the reality of network failures in a distributed
environment, which is encapsulated by the CAP Theorem.
The Goal: ACID Properties
Traditional relational databases are designed around ACID to ensure data reliability for
transactional operations:
Atomicity: All operations within a transaction succeed, or the entire transaction is rolled back.
It's an all-or-nothing proposition.
DISTRIBUTED ACID
SYSTEMS:
Consistency (ACID): A transaction moves the database from one
valid state to another, upholding all defined integrity rules
(constraints, triggers).
Isolation: Concurrent transactions execute as if they were running
serially. One transaction's intermediate state is hidden from
others.
Durability: Once a transaction is committed, its changes are
permanent, surviving system failures.
In a distributed system, achieving these properties across
multiple nodes is difficult, primarily due to network latency and the
potential for partitions.
Distributed systems come in varying shapes, sizes, and forms but
they all have a few typical characteristics and are exposed to
similar complications. As distributed systems get larger and more
spread out, the complications get more challenging. Added to
that, if the system needs to be highly available the challenges
only get multiplied.
Consistency:

• Consistency is not a very well-defined term but in the context of CAP it alludes to atomicity and isolation.
Consistency means consistent reads and writes so that concurrent operations see the same valid and
consistent data state, which at minimum means no stale data.
• In ACID, consistency means that data that does not satisfy predefined constraints is not persisted. That’s
not the same as the consistency in CAP.
• Brewer’s Theorem was conjectured by Eric Brewer and presented by him as a keynote address at the
ACM Symposium on the Principles of Distributed Computing (PODC) in 2000. Brewer’s ideas on CAP
developed as a part of his work at UC Berkeley and at Inktomi. In 2002, Seth Gilbert and Nancy Lynch
proved Brewer’s conjecture and hence it’s now referred to as Brewer’s Theorem (and sometimes as
Brewer’s CAP Theorem). In Gilbert and Lynch’s proof, consistency is considered as atomicity. Gilbert and
Lynch’s proof is available as a published paper titled “Brewer’s Conjecture and the Feasibility of
Consistent, Available, Partition-Tolerant Web Services”
• In a single-node situation, consistency can be achieved using the database ACID semantics but things get
complicated as the system is scaled out and distributed.
Availability:

• Availability means the system is available to serve at the time when it’s needed. As a
corollary, a system that is busy, uncommunicative, or unresponsive when accessed is not
available.
• Some, especially those who try to refute the CAP Theorem and its importance, argue that a
system with minor delays or minimal hold-up is still an available system.
• Nevertheless, in terms of CAP the defi nition is not ambiguous; if a system is not available
to serve a request at the very moment it’s needed, it’s not available. That said, many
applications could compromise on availability and that is a possible trade-off choice they
can make.
Partition Tolerance:
• Partition Tolerance (P) is the mandatory pillar of the CAP theorem chosen by modern distributed systems
due to the shift from vertical scaling (scaling up) to horizontal scaling (scaling out).
• Horizontal scaling—building a cluster from many commodity hardware units—is now the preferred, cost-
effective model for achieving high performance and scalability, replacing the expensive and limited
approach of building monolithic, powerful supercomputers. Because this chosen path inherently involves
distributing data across multiple independent nodes, network faults and partitions are inevitable
realities. Partition Tolerance, therefore, measures a system's ability to remain operational and continue
servicing requests even when some of its cluster members become temporarily unreachable or unavailable
due to network issues. The very nature of a distributed cluster built on commodity hardware, which is
susceptible to the Fallacies of Distributed Computing (such as assuming the network is reliable or
latency is zero), mandates that the system must be designed to tolerate and manage these faults to
remain effective.

• The historical challenge to adopting horizontal scaling was the traditional emphasis on strong Consistency
(C), which architects felt was compromised by distributed environments. However, the constraints and high
costs of vertical scaling eventually pushed the industry toward distributing data. Vertical scaling is limited
by vendor lock-in, significantly higher costs, and a finite data growth perimeter (the largest machine still
has a limit). Furthermore, vertical scaling requires complex proactive provisioning to budget for future
large-scale growth, which is often impossible to predict accurately. Horizontal scaling, while introducing the
risks detailed in the Fallacies of Distributed Computing, overcomes these vertical scaling limitations,
making Partition Tolerance the required cornerstone for any modern, scalable, and cost-efficient cloud
architecture.
UPHOLDING CAP:
• Achieving consistency, availability, and partition tolerance at all times in a large distributed system is not possible
and Brewer’s Theorem already states that. Gilbert and Lynch’s proof to delve deeper into how and why Brewer is
correct.
• The scenario where a Traditional transactional RDBMS (Relational Database Management System) chooses
Option 1 (Consistency and Partition Tolerance over Availability) and its availability is affected by hardware
failure leading to partitioning can be explained in detail by looking at the implications of the CAP theorem and
the nature of RDBMS:
CAP Theorem Context for RDBMS (Option 1)
• Traditional RDBMS systems are fundamentally designed to uphold the ACID properties (Atomicity,
Consistency, Isolation, Durability), with Consistency (C) being their highest priority.
• When a traditional RDBMS is horizontally scaled (distributed across multiple nodes), and a network partition (P)
occurs—like a hardware failure—the system is forced to choose between Consistency (C) and Availability (A),
as dictated by the CAP Theorem.
• Partition Tolerance (P) is unavoidable: In a distributed system, the possibility of network or node failure
(partitioning) must be assumed.
• RDBMS prioritizes Consistency (C): To ensure that every read operation returns the most recent, correct, and
committed data (strong consistency), the RDBMS will sacrifice availability during a partition.
• Option 1 = CP System: The RDBMS chooses Consistency (C) and Partition Tolerance (P), which means
Availability (A) is compromised.
 UPHOLDING CAP:
Hardware Failure Leading to Partitioning
A hardware failure that leads to a partitioning event is a concrete manifestation of the abstract "P" in
the CAP theorem.
1. The Partition Event
•Hardware Failure: A server's network card fails, a switch goes down, a cable is severed, or even a
database node itself crashes.
•Resulting Partition: This hardware issue prevents the now-isolated nodes from communicating and
synchronizing data, effectively dividing the system into separate sub-networks (partitions). For instance,
in the scenario described in Figure 9-4, the "replication failure due to network unavailability" is a type of
partition.
2. The Consistency Mandate
•The RDBMS has to ensure that no transaction violates data integrity. If a write operation $\text{A}$
completes on node $\text{X}$, but the partition prevents it from reaching node $\text{X}'$, then $\text{X}$
and $\text{X}'$ hold inconsistent data ( $\text{v1}$ on $\text{X}$, $\text{v0}$ on $\text{X}'$).
3. Compromising Availability
To guarantee that a subsequent read operation $\text{B}$ always sees the consistent (latest) value
$\text{v1}$ (as shown in Figure 9-5), the RDBMS must stop all operations that rely on the affected data
until the partition is resolved and the data is synchronized.
The system will take the following actions, leading to compromised availability:
•Blocking Reads: Any read request (like $\text{B}$'s read) targeting an unsynchronized or partitioned
node will be blocked or denied a response. The system waits for the partitioned nodes to re-establish
communication and perform the necessary synchronization (e.g., using a distributed commit protocol like
Two-Phase Commit - 2PC).
UPHOLDING CAP:
•Blocking Writes: New write requests may also be blocked
until all replicas acknowledge the update to ensure it can be
committed consistently across all nodes. If the failure occurs
before all nodes confirm the write, the entire transaction is
often rolled back.
•Failure of a Master Node: In a master-replica setup, if the
primary (master) node fails (a type of partition), no new writes
can be processed, and a time-consuming failover process to
elect a new master is required, during which the system is
unavailable for writes.
In essence, the choice to prefer Consistency (C) and
Partition Tolerance (P) means that when a hardware-
induced partition occurs, the system's response is to halt
operations (sacrificing Availability) rather than risk serving
stale or incorrect data.
Comparison with Option 3 (AP)
In contrast, a system choosing Option 3 (Availability and
Partition Tolerance over Consistency)—often seen in some
NoSQL databases—would, during a hardware-induced
partition, allow both the write on $\text{X}$ and the read on
$\text{X}'$ to proceed. The read would get the stale value
($\text{v0}$), thus compromising consistency, but the system
remains fully available. This is often referred to as Eventual
Consistency.
UPHOLDING CAP:
UPHOLDING CAP:
 • The CAP theorem (also called Brewer’s
The CAP theorem theorem) says that in a distributed database
(also called Brewer’s system, you can only guarantee two out of the

theorem) following three properties at the same time:

• In a distributed system, network failures are
inevitable — so Partition Tolerance (P) is non-
negotiable in real-world systems.
Property Meaning • So practically, designers must choose between:
Every node sees the same
• CP (Consistency + Partition tolerance)
data at the same time. (After a
C - Consistency
write, all reads return the latest • AP (Availability + Partition tolerance)
value.)
Every request receives a
response, even if some nodes
A - Availability
have failed. (The system
remains usable.)
The system continues to
work even if communication
P - Partition Tolerance
between nodes breaks
(network partition occurs).
Compromising on Availability:

• Sacrifice availability to maintain consistency when a network partition happens.

• If part of the cluster is unreachable, the system refuses to serve some requests (especially writes or reads) until it can ensure consistency
across all nodes.
Example
• MongoDB (in replica set mode) and HBase are CP systems.
• If the primary node fails and no new primary is elected yet:
• The database won’t accept writes (unavailable temporarily),
• This ensures no inconsistent writes happen.
Advantages
• Data is always consistent.
• Applications don’t see stale or conflicting data.
Disadvantages
• Some parts of the system may become unavailable during network issues.
• User experience may suffer (“Service temporarily unavailable”).
Compromising on Partition Tolerance
(Choosing CA System)

• Sacrifice partition tolerance, assuming network failures never happen or are rare.

• The system ensures:

• Every node is consistent (C)

• Always available (A) but if a network partition happens, the system fails or behaves unpredictably.

Example

• Traditional single-node databases like MySQL or PostgreSQL (non-clustered) are CA systems.

• Since there’s only one node (or a local cluster with reliable LAN), partitioning is not a major concern.

Advantages

• Simple to design.

• Always consistent and available when network is stable.

Disadvantages

• Cannot scale easily across distributed regions.

• Fails or freezes under network partition — not suitable for large distributed systems.
Compromising on Consistency (Choosing AP System)

• Sacrifice strict consistency to keep the system available and tolerant to partitions.

• Even if nodes can’t talk to each other:

• The system continues to serve requests.

• But different nodes might return different (outdated) data temporarily.

• Later, the system will synchronize to achieve eventual consistency.

Example

• Cassandra, DynamoDB, CouchDB, and Riak are AP systems.

• If one node can’t reach another:

• Each continues to accept writes,

• When connection restores, they resolve conflicts (via timestamps, version vectors, etc.).

Advantages

• High availability even during network failures.

• Excellent for globally distributed systems.

Disadvantages

• Reads might return stale data.

• Conflict resolution can be complex.

 Summary Table

Type Compromises Guarantees Example Systems Use Case

Financial
Consistency + MongoDB (Replica
CP Availability transactions,
Partition Tolerance Set), HBase
Banking
Cassandra,
Availability + Social media, IoT,
AP Consistency DynamoDB,
Partition Tolerance Analytics
CouchDB
MySQL,
Consistency + Local applications,
CA Partition Tolerance PostgreSQL
Availability single datacenter
(Single Node)
Real-World Analogy
Imagine a group
chat system split across two regions due to a network issue.
•CP system: Some users can’t send messages until the network is fixed (consistent chat history, but
unavailable).
•AP system: All users can send messages, but some may see delayed or out-of-order messages (available,
but inconsistent).
•CA system: The chat works only if there’s no network partition — otherwise, it fails.
CONSISTENCY IMPLEMENTATIONS IN A FEW NOSQL
PRODUCTS
1) Distributed consistency in MongoDB (replica sets & sharded clusters):
Core mechanisms
• Replica sets: MongoDB uses replica sets (one primary, multiple secondaries). Writes go to the primary and are replicated to
secondaries. Reads can be served from primary or (optionally) from secondaries.
• Write concern: controls durability/acknowledgement semantics for writes (e.g., w:1, w:majority, or w:N). Choosing w:majority
gives stronger guarantees that a write is persisted on a majority of voting members.
• Read concern: controls how up-to-date/read-isolation reads are (e.g., local, majority, linearizable in recent versions). majority
readConcern ensures you only see writes acknowledged by a majority.
• Causal consistency (sessions): MongoDB supports causal consistency for client sessions so that reads reflect causally prior
writes when configured. This uses combinations of read/write concern + logical clocks.
Behavior under partition / failure
• MongoDB is typically considered CP (Consistency + Partition tolerance) in CAP tradeoffs when you configure reads/writes for
majority and disallow accepting writes on isolated primaries. In practice you can tune toward availability (e.g., allow reads from
secondaries) or toward stronger consistency (use w:majority and readConcern: majority). During elections or partitions, writes may
be refused until a primary is elected (temporary unavailability if you insist on majority).
Practical notes
• If you need strong, predictable reads after writes, use writeConcern: majority plus readConcern: majority (or use causal
sessions). That gives you a strong guarantee that a subsequent read will see the write. If you relax these (reads from secondaries,
w:1), you can see stale data.
CONSISTENCY IMPLEMENTATIONS IN A FEW
NOSQL PRODUCTS:
2) Eventual consistency in CouchDB
Core mechanisms
• MVCC & append-only documents: CouchDB stores documents with MVCC-style revisions; each change creates a new revision
rather than overwriting the old. This simplifies local writes and concurrent updates.
• Incremental replication: CouchDB’s replication is incremental and multi-master: independent CouchDB nodes can accept writes
locally and later synchronize by pushing/pulling changes between databases. This is the basis of CouchDB’s AP (available +
partition-tolerant) design and eventual consistency guarantee.
• Conflict detection & resolution: When two replicas change the same document concurrently, CouchDB marks the document as
in conflict. It preserves conflicting revisions and exposes conflicts to the application; the app (or a user-defined policy) chooses
which revision becomes the winning/merged revision. CouchDB does not automatically force one canonical resolution for arbitrary
document merges.
Behavior under partition / failure
• Because each node can accept writes independently, CouchDB stays available during partitions. When the network heals,
replication exchanges changes and conflicts are detected and surfaced for resolution (eventual convergence). This makes
CouchDB a classic AP system in CAP terms.
Practical notes
• CouchDB is excellent for disconnected/edge/mobile scenarios where nodes must accept writes offline and sync later. But you
must design conflict-resolution strategies (application-level merge, last-writer-wins if acceptable, or manual resolution workflows).
CONSISTENCY IMPLEMENTATIONS IN A FEW
NOSQL PRODUCTS:
2) Eventual consistency in CouchDB
CONSISTENCY IMPLEMENTATIONS IN A FEW
NOSQL PRODUCTS:
3) Eventual / tunable consistency in Apache Cassandra
Core mechanisms
• Dynamo-inspired architecture: Cassandra is modeled after Dynamo — decentralized ring, partitioning, replication across nodes.
It exposes tunable consistency per operation by letting clients pick consistency levels (e.g., ONE, QUORUM, ALL).
• Consistency levels & R + W > N rule: If the sum of replicas required for a read (R) and a write (W) is greater than the replication
factor (N), you can get strong consistency for that operation (practically read-your-writes / linearizable-like guarantees for that
operation). Otherwise you get weaker (eventual) guarantees. This allows per-operation tradeoffs between latency/availability and
consistency.
• Mechanisms to converge: Cassandra uses hinted handoff, read repair, and anti-entropy (Merkle tree) repairs to bring
replicas into consistency over time. For stronger atomic consistency on conditional updates, Cassandra supports lightweight
transactions (LWT) implemented via Paxos.
Behavior under partition / failure
• Cassandra is designed to maximize availability and partition tolerance (AP by default), but clients can choose higher
consistency levels for specific reads/writes when needed. For global clusters, network partitions lead to nodes accepting writes
locally; eventual convergence is achieved later. Apache
Practical notes
• Use low consistency levels (e.g., ONE) for low-latency reads/writes and high availability, accept stale reads. Use QUORUM or
LOCAL_QUORUM (with appropriate replication) for stronger guarantees that balance latency and correctness. For strict compare-
and-set semantics, use LWT/Paxos.
CONSISTENCY IMPLEMENTATIONS IN A FEW
NOSQL PRODUCTS:
4) Consistency in Membase (historical) → Couchbase
Context / history
• Membase was a distributed key-value store (an evolution of memcached with persistence and clustering). Over time Membase
and CouchDB technologies converged and the product line evolved into Couchbase Server. Today you should read Couchbase
docs for modern behavior, but understanding Membase helps explain Couchbase’s key-value origins.
Consistency model (Couchbase / Membase lineage)
• Key-value document reads are strong/consistent for the active (primary) vBucket: Couchbase typically serves key reads
from the active vBucket owner for that key, so a client reading a key gets the most recent value from the active node (strong
consistency for direct key operations). Query/index views may be eventually consistent depending on indexing and whether they
use background indexers or global replication (XDCR).
• XDCR (cross datacenter replication): Replication across clusters (XDCR) is typically eventual, so writes replicated cross-
datacenter converge asynchronously.
Practical notes
• Couchbase (Membase heritage) gives low-latency strongly consistent key-value operations within a cluster node owning the
active partition, while higher-level services (views, indexes, XDCR) introduce eventual semantics. If your workload is key-value
retrievals and you target the active node, you get strong consistency; if you rely on global replication or secondary indexes, plan
for eventual behavior.
PERFORMANCE TUNING
Goals of Parallel Algorithms (performance-tuning view):
Objective: get more useful work done per second and/or reduce time per task with minimal cost (energy,
money, complexity).
Concrete goals when tuning for performance:
• Minimize wall-clock time for a workload (latency).
• Maximize throughput (tasks completed per second).
• Maintain/raise efficiency (use hardware effectively).
• Scale predictably as you add resources (linear or graceful sublinear).
• Keep results correct (avoid race conditions, maintain numerical stability).
• Control costs: energy, money, or resource usage.
Key metrics to watch:
T₁ (work on 1 core), Tₚ (time on p cores), throughput λ (tasks/sec), latency W (sec/task), utilization,
efficiency.
Implications of Reducing Latency (practical):
• Latency = time to complete one request/job. Reducing latency affects the system in predictable ways:
• User experience — immediate effect; perceived speed improves.
• Throughput — if workers are independent, throughput ≈ 1/latency per worker; reducing latency increases
throughput.
• Resource balancing — shorter latency can expose other bottlenecks (I/O, locks).
• Cost tradeoffs — reducing latency often requires more CPU, memory, or specialized hardware (SSD,
NVMe, RDMA).
• Example (digit-by-digit arithmetic):
If one request takes 20 ms = 0.020 s, throughput per worker = 1 / 0.020 = 50.
If you reduce latency to 10 ms = 0.010 s, throughput per worker = 1 / 0.010 = 100.
Calculation: 1 ÷ 0.020 = 50 (since 0.02 × 50 = 1). 1 ÷ 0.01 = 100.
Tuning actions to reduce latency (single request):
• remove blocking I/O (use async, prefetching)
• reduce syscalls and context switches
• optimize hot loops (vectorize, unroll)
• ensure data fits in cache (locality)
• reduce lock contention (lock-free, sharding)
How to Increase Throughput (engineering patterns)
• Throughput = number of tasks completed per unit time (λ).
Common strategies:
• Increase parallelism (add workers/cores/threads) — careful with contention and overhead.
• Decrease per-task latency — reduces time each worker is busy.
• Pipeline stages — turn tasks into stages so many tasks are in flight (assembly line).
• Batching — process many items at once to amortize overhead (e.g., syscalls, kernel launches).
• Vectorization / SIMD — process multiple data items per instruction.
• Offload to accelerators — GPUs for dense arithmetic, FPGAs for special kernels.
• Asynchronous I/O and concurrency — avoid blocking threads.
• Load balancing — avoid idle resources.
• Caching & Memoization — avoid rework.
• Throughput example:
Single worker does 10 tasks/sec. 10 workers ideally do 100 tasks/sec. If contention/overhead reduces per-
worker effective rate to 9 tasks/sec, total = 10 × 9 = 90 tasks/sec. So overhead matters.
Linear Scalability — what it means and why it breaks
• Linear scalability: doubling resources ⇒ doubling throughput (or halving latency) — ideal case.
• Why it fails in practice:
• Amdahl’s Law (sequential fraction)
• Communication and synchronization costs (network latency, barriers)
• Resource contention (memory bandwidth, locks)
• Load imbalance (some workers idle)
• Overheads (thread management, context switching)
• Visual (ASCII) — ideal vs actual speedup:
• Speedup
|
| Ideal: / (slope 1)
| /
|Real / (sub-linear due to overhead)
| /
+----------------- Number of processors
Influencing Equations:
• Influencing Equations — quick map
These are the core equations you’ll see over and over:
1
• Amdahl’s law: 𝑆 𝑝 = 1−𝑠
𝑠+ 𝑝

• Gustafson’s law: 𝑆 𝑝 = 𝑝 − 𝑠 𝑝 − 1
• Little’s law: 𝐿 = 𝜆𝑊
𝑇
• Work-Span bound: 𝑇𝑝 ≥ max ⁣ ( 1 , 𝑇∞ )
𝑝

• Message (cost) model: 𝑇𝑚𝑠𝑔 = 𝛼 + 𝛽𝑛)or extended: 𝛼 + 𝛽𝑛 + 𝛾𝑛𝑝𝑟𝑜𝑐 (

message size
• Communication time (bandwidth+latency): 𝑇𝑐𝑜𝑚𝑚 = latency +
bandwidth
1) Amdahl’s Law (limits of parallelism)
1
• Equation: 𝑆 𝑝 = 1−𝑠
𝑠+
𝑝

• 𝑠 =fraction strictly sequential (0..1).

• 𝑝= number of processors.
• As 𝑝 → ∞, 𝑆 𝑝 → 1/𝑠 —the sequential fraction caps speedup.
Worked example (digit-by-digit):
Let 𝑠 = 0.10(10% sequential), 𝑝 = 8.
Compute parallel fraction: 1 − 𝑠 = 0.90.
1−𝑠 0.90
Compute = .
𝑝 8
Step: 0.90 ÷ 8 = 0.1125.
1−𝑠
Now denominator: 𝑠 + = 0.10 + 0.1125 = 0.2125.
𝑝
Speedup: 𝑆 8 = 1 ÷ 0.2125 ≈ 4.7058823529..×4.706 ≈
Interpretation & tuning:
• If you can reduce 𝑠from 0.10 to 0.02, the max speedup increases dramatically. Prioritize removing or parallelizing sequential
hotspots.
• Amdahl is pessimistic for scaled problems (when parallel work grows with problem size); use Gustafson there.
Short exam answer: Amdahl’s law gives the theoretical upper bound of speedup for a fixed problem size, determined by the
sequential fraction.
2) Little’s Law (queueing baseline)
Equation: 𝐿 = 𝜆𝑊
• 𝐿= average number of jobs in the system (concurrency).
• 𝜆= throughput (jobs/sec).
• 𝑊= average latency (sec/job).
Worked example:
Target throughput 𝜆 = 200jobs/sec. Desired latency 𝑊 = 50ms = 0.050s.
Compute 𝐿 = 𝜆 × 𝑊 = 200 × 0.050.
Step: 200 × 0.050 = 10.000.
So 𝐿 = 10concurrent jobs needed (on average).
Interpretation & tuning:
• If measured concurrency >Little’s law prediction, you may have queuing or blocking.
• Use Little’s Law to size thread pools, batch sizes, and to set concurrency controls.
Short exam answer: Little’s Law relates concurrency, throughput and latency and is used to size systems.
3) Message Cost Model / Message Passing (α–β model)
Basic model:

• 𝑇𝑚𝑠𝑔 = 𝛼 + 𝛽𝑛

• 𝛼 =startup latency per message (ms) — fixed overhead (handshake, interrupt).

• 𝛽= per-byte transmission cost (ms/byte) — inverse of bandwidth.
• 𝑛= message size in bytes.
Extended variants: include serialization/deserialization cost, per-receiver cost in multicast, and contention factors.
Worked numeric example:
Assume 𝛼 = 0.50ms, 𝛽 = 0.01ms/byte (i.e., 100 KB/s? — these are illustrative), 𝑛 = 1000bytes.
Compute 𝛽𝑛 = 0.01 × [Link]: 0.01 × 1000 = [Link].
Then 𝑇𝑚𝑠𝑔 = 𝛼 + 𝛽𝑛 = 0.50 + 10.000 = 10.500ms.
So sending a 1 KB message costs 10.5 ms.
Tuning implications:
• For small messages (𝑛tiny ) ,𝛼dominates → batch small messages.
• For large 𝑛, per-byte term dominates → compress or use RDMA/high-bandwidth links.
• Use asynchronous transfers and overlap computation with communication (nonblocking sends).
Short exam answer: The α-β model separates latency and bandwidth effects and guides batching vs compression
choices.
4) Partitioning (data & task partitioning)
• Goal: split work and data to minimize load imbalance and communication.
• Common strategies:
• Block (contiguous) partitioning: divide data into contiguous chunks (good for locality).
• Cyclic partitioning: assign items round-robin (pointer chasing or irregular access patterns).
• Range / hash partitioning: used for distributed key-based systems (DBs/MapReduce).
• Owner computes rule: place data on the node that will process it to avoid network transfers.
Metrics:
max 𝑖 𝐿𝑖
• Load imbalance ratio: 𝐼 = 1 where 𝐿𝑖 is work on node 𝑖. Ideal 𝐼 = 1.
෌𝑖 𝐿𝑖
𝑝

• Communication volume: total bytes moved between partitions.

• Worked example (imbalance): 3 workers with loads [120, 100, 80] units.
Average load = 120 + 100 + 80 ÷ [Link]: 300 ÷ 3 = 100.
Max load = 120. Imbalance 𝐼 = 120 ÷ 100 = 1.2.→ 20% imbalance.
Tuning tips:
• Balance compute, minimize edge cuts (graph partitioning: METIS, Scotch).
• For skewed keys, use skew mitigation (splitting hot keys, sampling).
• Favor partitions that reduce communication (locality).
• Short exam answer: Partitioning trades off load balance vs communication—pick a scheme that minimizes total cost
(compute+comm).
Scheduling in Heterogeneous Environments
• Problem: nodes have different speeds, memory, accelerators; tasks vary in cost.
• Models & heuristics:
• Weighted round-robin / capacity-based: allocate work proportional to node speed. If node A speed=1, B speed=2, allocate twice
to B.
• HEFT (Heterogeneous Earliest Finish Time): a DAG scheduler that assigns tasks to minimize finish time using estimated
execution & communication costs.
• Min-min / Max-min heuristics: assign smallest/longest tasks to fastest nodes to minimize makespan.
• Work stealing with weights: idle fast nodes steal proportionally.
Worked allocation example: two machines M1 speed = 1unit/sec, M2 speed = 2units/sec. You have 90 units of identical work. How
to split proportionally?
Total speed = 1 + 2 = 3.
M1 share = 1/3of work → 1/3 × [Link]: 90 ÷ 3 = [Link] M1 gets 30 units.
M2 share = 2/3 × [Link]: 90 ÷ 3 = 30, then 30 × 2 = [Link] M2 gets 60 units.
Time: M1 finishes in 30 ÷ 1 = 30s; M2 finishes in 60 ÷ 2 = 30s → balanced.
Tuning tips:
• Use online profiling to estimate effective node speed.
• Use task size variability aware scheduling (chunking large tasks).
• Affinity: bind tasks to nodes with required data/accelerators.
• Short exam answer: Schedule proportional to node capacity; use HEFT for DAGs or adaptive load-balancing for streaming tasks.
Additional Map-Reduce Tuning (practical knobs)
Key knobs and their effects:
• Input split size (map tasks count): smaller splits → more mappers → more parallelism, but more overhead.
• Number of reducers: too few → bottleneck; too many → overhead and many small outputs.
• Combiner: local reduce to cut shuffle volume (use when associative+commutative).
• Speculative execution: duplicate slow tasks to avoid stragglers (costly if cluster busy).
• Spill buffer / memory tuning: tune map buffer to reduce spills to disk.
• Compression: compress map outputs to reduce network shuffle (CPU vs IO tradeoff).
• Locality & data placement: schedule mappers where data resides.
• Task JVM reuse: avoid VM startup costs.
• Numeric tradeoff example (map output compression):
If uncompressed map output = 100 GB, network bandwidth effective = 2 GB/s, sending takes 100 ÷ 2 = 50s. If
compression ratio = 4× (i.e., compressed size = 25 GB) and compression+decompress CPU cost = 10 s total,
then total = 25 ÷ 2 + 10 = 12.5 + 10 = 22.5s → better than 50 s. Show steps: 100 ÷ 4 = 25GB compressed; 25 ÷
2 = 12.5s; +10 s CPU = 22.5 s.
• Tuning tips: choose compression codec (Snappy for speed, gzip for better ratio), set [Link],
and tune mapper buffer thresholds to reduce spills.
• Short exam answer: MapReduce tuning balances map/reduce parallelism, memory, IO, and network (use
combiner, tune splits and reducers, use compression to reduce shuffle).
Communication Overheads (types & modeling)
• Sources of overhead:
• Latency (startup cost) — per message.
• Bandwidth limitations — per-byte cost.
• Serialization/deserialization — CPU overhead.
• Contention & congestion — multiple flows share links/switches.
• Synchronization/barriers — global waits (collectives, allreduce).
• Protocol overheads — TCP slow start, handshakes.
Model (practical):
• 𝑇𝑐𝑜𝑚𝑚 = 𝛼 + 𝛽𝑛 + 𝑇𝑠𝑒𝑟𝑖𝑎𝑙𝑖𝑧𝑒 + 𝑇𝑞𝑢𝑒𝑢𝑒_𝑤𝑎𝑖𝑡 + 𝑇𝑠𝑦𝑛𝑐
• Where 𝑇𝑠𝑒𝑟𝑖𝑎𝑙𝑖𝑧𝑒 often proportional to message size (or CPU-bound constant).
• Example: allreduce on N nodes—costs include log(N) rounds (tree algorithm) or O(N) depending on
algorithm. Design choice matters.
• Mitigations: asynchronous comms, overlap compute/comm, use collective algorithms tuned for topology,
offload to NICs (RDMA/GPUDirect).
• Short exam answer: Communication overheads include fixed startup, per-byte costs, and synchronization
delays; mitigate via batching, compression, and overlapping.
Compression (tradeoffs & choices)
• Why: reduce network and disk I/O at expense of CPU. Key trade: CPU time vs I/O time.
Common codecs:
• Snappy — very fast, moderate compression ratio (good for map output).
• LZ4 — similar use case.
• gzip (deflate) — higher ratio, slower.
• Zstd — good compromise (fast + good ratio).
• bzip2 — high compression, slow.
• When to use: compression helps when I/O/bandwidth is the bottleneck and CPU slack exists.
• Numeric example: previously shown map-shuffle example demonstrates savings.
• Tuning tips: measure compression ratio and compression CPU cost; choose codec that yields
net latency reduction.
• Short exam answer: Compression reduces I/O cost but consumes CPU; pick codec based on
network vs CPU balance.
File Block Size (HDFS / distributed files)
• Effect of block size (B):
• Metadata overhead: smaller B → more blocks → larger NameNode metadata.
• I/O throughput: large B reduces per-block seek overhead and increases sequential
throughput; good for big files.
• Small files problem: many small files cause NameNode overload and poor throughput.
• Numeric example — blocks in 1 TB:
1 TB = 1024GB = 1024 × 1024MB = 1,048,576MB. Default HDFS block = 128MB.
Compute number of blocks: 1,048,576 ÷ [Link]: 1,048,576 ÷ 128 = [Link] 8192
blocks for 1 TB at 128 MB blocks.
• Tuning tips: use larger block size for big files (256MB/512MB) to reduce RPC overhead.
For many small files, pack them (HAR, SequenceFile) or use object stores.
• Short exam answer: Block size trades metadata overhead and map task count vs
sequential throughput—pick larger blocks for large files, avoid many tiny blocks.
Parallel Copying (distcp, multithreaded, pipelined)
• Goals: copy large datasets across nodes/clusters quickly and reliably.
• Common tools & approaches:
• DistCp (Hadoop): map tasks copy files in parallel; uses MapReduce to parallelize and retry.
• rclone / rsync with multiple streams: parallelize file transfers.
• Multi-stream TCP: open multiple connections per file to increase throughput.
• Pipelined copy: start transfer of chunk while creating next chunk.
• Checksums & verification: ensure data integrity (MD5, CRC).
• Tuning knobs: number of parallel mappers, per-map concurrency, buffer sizes, checksum offload,
compression for network.
• Numeric example: speed scaling with parallelism:
Single stream sustained throughput = 100 MB/s. Copy size = 1 TB = 1024 × 1024MB = 1,048,576 MB.
Single stream time = 1,048,576 ÷ [Link]: 1,048,576 ÷ 100 = 10,485.76s ≈ 174.76 minutes.
With 8 parallel streams each 100 MB/s aggregate = 800 MB/s (if no contention): time = 1,048,576 ÷
[Link]: 1,048,576 ÷ 800 = 1,310.72s ≈ 21.845 minutes. So ~8× speedup if network and disks can
sustain it.
• Caveat: parallel streams can cause congestion; measure end-to-end.
• Short exam answer: Parallel copying splits dataset across multiple concurrent transfers (map tasks or
streams) to increase throughput; tune degree of parallelism to avoid network/disk contention.
Isolation Strategies In a Distributed System, data is spread across
multiple machines or nodes. Ensuring the ACID
in Distributed ACID properties, especially Isolation and Atomicity (often

Systems combined as "Distributed Transactions"), becomes

significantly more complex due to network latency,
node failures, and the need for coordination across
independent machines.
The CAP Theorem suggests that a distributed
system can only guarantee two out of three
properties: Consistency, Availability, and Partition
tolerance. Distributed ACID systems, by prioritizing
Consistency (C) and Partition tolerance (P), often
sacrifice Availability (A) during network failures,
aligning with the principles of Strong Consistency.
SCENARIOS: Banking Transaction Problem Statement
Design a banking transaction system where simultaneous
withdrawals and deposits occur on customer accounts. Ensure that
transactions either fully complete or roll back on failure to prevent
inconsistencies, such as negative balances or lost updates, when
multiple users operate at the same timE,.
Flash Sale Inventory Problem Statement
Create an inventory management system that correctly tracks stock
levels during flash sales with thousands of concurrent customers.
The system must prevent overselling by ensuring stock counts are
repeatably read and correctly updated as orders are placed,
handling simultaneous accesses and updates without race
conditions or lost updates.
Flight Booking Problem Statement
Develop a distributed flight booking system that allows users
worldwide to book seats in real time. The system must prevent
double booking in cases of high concurrency, using strict
serializable isolation, while ensuring that each seat is reserved only
once regardless of the number of simultaneous booking attempts.
SCENARIOS:

Healthcare Records Problem Statement

Implement a healthcare record system that supports multiple concurrent updates and reads
across distributed clinics. Ensure that each patient's information remains accurate, allowing only
committed transactions to be visible and preventing dirty or unrepeatable reads that could lead to
medical errors.
Real-Time Warehouse Analytics Problem Statement
Build a real-time data warehouse supporting instant analytics on incoming shipments and orders
from distributed warehouses. The system must synchronize updates, avoiding unrepeatable
reads or phantom records while efficiently processing analytics queries that aggregate rapidly
changing data.
E-Commerce Order Problem Statement
Design a distributed, cloud-scale e-commerce platform to support thousands of concurrent users
placing orders. Use multi-version concurrency control and locking strategies to ensure that each
product stock level and transaction history is correctly updated, preventing lost updates and
maintaining atomicity and consistency across all nodes.