Distributed Systems for Data Engineers

What this module covers and what it deliberately does not:This module covers the concepts a data engineer needs to make correct architectural decisions and debug distributed failures — CAP theorem, consistency models, replication, partitioning, joins, fault tolerance, and consensus. It does not cover implementing a distributed database from scratch, the full Paxos or Raft algorithms in academic depth, or network programming. Those are software engineering and systems programming topics. If you want those, read Designing Data-Intensive Applications by Martin Kleppmann — it is the best book on this subject ever written.

The real insight of CAP: In any real distributed system, network partitions will happen. A cable gets unplugged. A switch fails. An AWS availability zone loses connectivity for 4 minutes. You cannot prevent partitions — you can only choose how your system behaves when they occur. This means Partition Tolerance is not optional. The real trade-off is between Consistency and Availability during a partition.

During a partition, you must choose: do you refuse to respond until the partition heals and you can guarantee a consistent answer (CP — prefer consistency over availability), or do you respond with potentially stale data and risk returning an inconsistent answer (AP — prefer availability over consistency)?

# Scenario: PhonePe user tops up wallet. You read the balance immediately after.

# With STRONG CONSISTENCY (linearisable):
write_wallet_balance(user_id='U1234', new_balance=500_00)  # ₹500 in paise
balance = read_wallet_balance(user_id='U1234')
# balance is GUARANTEED to be 500_00
# The read was routed to any node — all nodes agree because write was fully replicated

# With EVENTUAL CONSISTENCY:
write_wallet_balance(user_id='U1234', new_balance=500_00)
balance = read_wallet_balance(user_id='U1234')
# balance MIGHT be 500_00 (if read hits the node that just received the write)
# balance MIGHT be 350_00 (old value — if read hits a replica not yet updated)
# Both are valid responses. The system makes no guarantee about which you get.

# In a financial system, this is catastrophic — showing the old balance
# after a top-up makes users think the top-up failed. They retry.
# Now you have a double top-up problem.

# → Financial data: always use strongly consistent reads
# → Analytics aggregations (eventual consistency of 30 seconds is fine):
#   Swiggy's live order count dashboard can be off by a few orders for 30 seconds.
#   Nobody is making financial decisions based on it.

# The read-your-writes workaround for eventually consistent stores:
# Route the user's own reads to the same node that accepted their write
# (sticky sessions / consistent routing) — a common pattern in Cassandra deployments

# PostgreSQL leader-follower replication lag scenario

# Leader receives a write: INSERT INTO orders VALUES (ORD-9999, ...)
# Leader writes to its WAL (Write-Ahead Log) — offset 10,441
# Leader acknowledges the write to the application
# Replication to follower begins asynchronously

# At this exact moment: replication lag = a few milliseconds
# Application immediately queries a follower for reporting:
# SELECT COUNT(*) FROM orders WHERE date = TODAY

# If the follower has not yet received offset 10,441:
# → COUNT returns a value that does not include ORD-9999
# → Report is stale by exactly 1 order
# → This is normal and expected with asynchronous replication

# How replication lag is measured (PostgreSQL):
# On the follower:
# SELECT now() - pg_last_xact_replay_timestamp() AS replication_lag;
# → "00:00:00.003" (3 milliseconds of lag — healthy)
# → "00:05:23" (5 minutes of lag — the follower is struggling, investigate)

# Common causes of high replication lag:
# 1. Follower disk I/O is saturated — cannot apply WAL fast enough
# 2. Long-running query on follower blocking WAL replay (PostgreSQL pre-14 issue)
# 3. Network congestion between leader and follower
# 4. Follower is under high read load, starving WAL replay threads

# Solutions:
# 1. Route time-sensitive reads to the leader, not followers
# 2. Use synchronous replication for critical data (blocks writes until follower confirms)
# 3. Monitor lag and alert on thresholds — 30s lag for analytics, 100ms for OLTP followers

Last-write-wins (LWW) is dangerous — it silently discards writes without errors. Two users simultaneously update a document. Both see a success confirmation. One update is silently lost. The user whose update lost has no idea. LWW is only safe when: the same user is the only writer to a key (impossible to guarantee in practice), or losing concurrent writes is acceptable (user preference settings, not financial data). CRDTs (Conflict-free Replicated Data Types) are the mathematically correct solution for data structures that must support multi-leader writes — but they are complex to implement and reason about.

# Cassandra cluster: N = 5 replicas (replication_factor = 5)
# Write consistency: W = 3 (quorum — majority)
# Read consistency:  R = 3 (quorum — majority)
# W + R = 6 > N = 5 → reads and writes overlap by at least 1 node
# → At least one node in every read quorum has the latest write
# → Strong consistency guaranteed

# But strong consistency means:
# Write must wait for 3/5 nodes to confirm → slower
# Read must contact 3/5 nodes and take the most recent value → slower
# If 3 nodes are unreachable, writes fail (W = 3 cannot be satisfied)

# Eventual consistency config (R = 1, W = 1):
# W + R = 2, not > N = 5 → no overlap guaranteed
# Write goes to 1 node, completes immediately
# Read goes to 1 node, returns immediately
# Maximum speed, minimum consistency

# FreshMart product catalogue (can tolerate slight staleness):
# N=3, W=1, R=1 → fastest, eventually consistent
# A product price being 30 seconds stale is acceptable

# FreshMart order status (must be accurate):
# N=3, W=2, R=2 → W+R=4 > N=3 → consistent
# A user checking whether their order is placed must get the truth

# Read repair — how leaderless systems eventually converge:
# Consumer reads from R=3 nodes. Gets values: v3, v3, v2 (one replica behind)
# System detects the stale replica (v2 while others have v3)
# Background read repair: sends the latest value (v3) to the stale replica
# Next read: all 3 nodes return v3 → consistent

# Hash partitioning: consistent_hash(key) % num_partitions → node assignment

# Good partition key: user_id (millions of distinct values)
# consistent_hash('user_U98765') % 32 = 17 → goes to node 17
# consistent_hash('user_U12345') % 32 = 3  → goes to node 3
# Distribution: roughly equal across all 32 nodes ✓

# Bad partition key: payment_method (4 distinct values: upi, card, netbanking, wallet)
# In India, UPI = 80% of all transactions
# consistent_hash('upi') % 32 = 9 → node 9 receives 80% of all traffic
# consistent_hash('card') % 32 = 22 → node 22 receives 15%
# Nodes 9 is a HOT SHARD — slower, more likely to fail, capacity bottleneck

# Bad partition key: timestamp (or anything that increases monotonically)
# All new writes go to the same partition (the latest time range)
# All reads for recent data also go to that partition
# This is called a HOT PARTITION — the most common mistake in time-series data

# Fix for time-series hot partitions:
# Compound partition key: (shard_id, timestamp)
# shard_id = hash(entity_id) % num_shards
# Writes are spread across shards, each shard has its own time ordering

# DynamoDB example for FreshMart orders:
# Bad:  partition_key = date          → all orders for today on one partition
# Good: partition_key = store_id      → 10 stores → 10 evenly distributed partitions
# Best: partition_key = store_id + "#" + hour  → 10 stores × 24 hours = 240 partitions

# Spark joining orders (100M rows, 50 partitions) with customers (10M rows, 50 partitions)
# JOIN condition: orders.customer_id = customers.customer_id

# Phase 1 — SHUFFLE (the expensive part):
# Both DataFrames are repartitioned by customer_id
# Each of the 50 executors scans its partition of orders
# For every row, it computes hash(customer_id) % 50 → target partition
# It SENDS that row over the network to the target executor
# Same for customers

# Network traffic during shuffle:
# orders:    100M rows × 500 bytes avg = 50 GB sent across the network
# customers: 10M rows × 200 bytes avg = 2 GB sent across the network
# Total: ~52 GB of network transfer just for this one JOIN

# Phase 2 — LOCAL JOIN (fast):
# Each executor now has all orders AND all customers for its hash range
# Joins them locally — no more network transfers

# Why shuffle joins kill performance:
# 1. Network transfer of 52 GB takes 5-10 minutes on typical cluster network
# 2. All executors must finish Phase 1 before Phase 2 can begin (barrier sync)
# 3. If one executor is slow (straggler), everything waits
# 4. If one customer_id is very common (data skew), one executor gets overloaded

# Optimisation: pre-partitioning
# If orders is ALREADY partitioned by customer_id (e.g., in Delta Lake with Z-order)
# Spark can skip shuffling orders — only customers needs to be shuffled
# 50 GB → 2 GB of network transfer. 25x faster.

# PySpark — explicit broadcast hint
from pyspark.sql.functions import broadcast

# store_metadata: 10 stores, tiny table (< 1 MB)
# orders: 500M rows, huge table
result = orders.join(
    broadcast(store_metadata),   # force broadcast — skip shuffle
    on='store_id',
    how='left'
)
# Spark sends store_metadata to all 200 executors once (10 stores × 200 bytes = trivial)
# Each executor joins locally — zero shuffle
# 500M rows processed with no network transfer for the join

# SQL equivalent (Spark SQL / Snowflake / BigQuery):
# SELECT /*+ BROADCAST(s) */ o.*, s.city, s.region
# FROM orders o
# JOIN store_metadata s ON o.store_id = s.store_id

# When NOT to broadcast:
# Table is > 500 MB → serialising and sending to 200 executors takes longer than the shuffle
# Memory on executors is tight → broadcasting a large table causes OOM
# The "small" table has data skew when joined → broadcast doesn't help with skew

# Rule of thumb: if the dimension/lookup table has fewer than 1M rows and fits in 500 MB,
# consider broadcasting. Always check the query plan to confirm it was applied.

# Problem: customer C00001 (Reliance Industries) has 10M orders out of 50M total
# After shuffle join, one executor gets 10M rows, others get ~800k
# Result: 1 executor runs for 45 minutes, others finish in 4 minutes

# Solution: SALTING — add randomness to the join key to spread the hot key

import random
from pyspark.sql.functions import col, expr, explode, array, lit, concat_ws

NUM_SALTS = 20  # must match on both sides

# Explode the small (or known hot-key) side: replicate each row NUM_SALTS times
store_salted = store_metadata.withColumn(
    'salt', explode(array([lit(i) for i in range(NUM_SALTS)]))
).withColumn('salted_key', concat_ws('_', col('store_id'), col('salt')))

# Add a random salt to the large side
orders_salted = orders.withColumn(
    'salt', (expr('rand()') * NUM_SALTS).cast('int')
).withColumn('salted_key', concat_ws('_', col('store_id'), col('salt')))

# Now join on salted_key instead of store_id
# C00001_0, C00001_1, ..., C00001_19 are spread across 20 different executors
result = orders_salted.join(store_salted, on='salted_key', how='left')

# Spark streaming checkpoint — saves state and progress to durable storage
spark = SparkSession.builder.getOrCreate()

query = (
    spark
    .readStream
    .format('kafka')
    .option('kafka.bootstrap.servers', 'broker:9092')
    .option('subscribe', 'freshmart.orders')
    .load()
    .writeStream
    .format('delta')
    .option('path', 'abfss://gold@stfreshmartdev.dfs.core.windows.net/orders_agg/')
    .option(
        'checkpointLocation',
        'abfss://checkpoints@stfreshmartdev.dfs.core.windows.net/orders_agg_ckpt/'
        # MUST be on durable storage (ADLS, S3) — not local disk
        # Local disk checkpoint = lost on executor failure = no recovery
    )
    .trigger(processingTime='1 minute')
    .start()
)

# What the checkpoint stores:
# 1. Kafka offsets consumed so far (per partition)
# 2. Streaming aggregation state (partial counts, sums, windows)
# 3. Schema of the output
# On restart, Spark reads the checkpoint and continues from exactly where it stopped
# No duplicate processing. No missed events.

# How often to checkpoint:
# More frequent = smaller recovery window, more overhead per checkpoint
# Less frequent = larger recovery window, less overhead
# Typical: every 1,000-10,000 events OR every 1-5 minutes for streaming jobs
# For batch jobs: checkpoint after each stage boundary — Spark does this automatically

# Non-idempotent (WRONG — unsafe after checkpoint restore):
def write_order_count(date: str, count: int):
    execute_sql(
        "UPDATE daily_stats SET order_count = order_count + %s WHERE date = %s",
        [count, date]
    )
    # If this runs twice (checkpoint restored after this line), count is doubled

# Idempotent (CORRECT — safe to run multiple times):
def write_order_count(date: str, count: int, pipeline_run_id: str):
    execute_sql("""
        INSERT INTO daily_stats (date, order_count, pipeline_run_id)
        VALUES (%s, %s, %s)
        ON CONFLICT (date) DO UPDATE
            SET order_count = EXCLUDED.order_count,
                pipeline_run_id = EXCLUDED.pipeline_run_id
        WHERE daily_stats.pipeline_run_id != EXCLUDED.pipeline_run_id
    """, [date, count, pipeline_run_id])
    # If this runs twice with the same pipeline_run_id:
    # The WHERE clause on pipeline_run_id prevents the second update from doing anything
    # Exactly-once semantics achieved through idempotent writes

# The pattern: make writes conditional on not having already applied this exact computation
# Mechanism: pipeline run ID, event ID, sequence number, or content hash as idempotency key

# Swiggy order placement — choreography-based saga
# Each service listens for events and publishes its own events
# No central coordinator

# Step 1 — Order Service
# Receives HTTP request to place order
# Creates order in PENDING state in its own database
# Publishes event: order.placement.requested

# Step 2 — Payment Service
# Listens for: order.placement.requested
# Attempts to deduct from wallet
# SUCCESS → publishes: payment.completed  (order_id, amount)
# FAILURE → publishes: payment.failed     (order_id, reason)

# Step 3 — Inventory Service
# Listens for: payment.completed
# Deducts item quantities from restaurant inventory
# SUCCESS → publishes: inventory.reserved  (order_id)
# FAILURE → publishes: inventory.failed    (order_id, reason)
#           → Compensating transaction: publish payment.refund.requested

# Step 4 — Order Service
# Listens for: inventory.reserved
# Updates order state to CONFIRMED
# Publishes: order.confirmed → triggers notification, logistics assignment

# ── Failure path — compensating transactions ─────────────────────────────────
# Inventory reservation fails AFTER payment was taken:

# Inventory Service publishes: inventory.failed
# Payment Service listens for: inventory.failed
# Payment Service REFUNDS the charge → publishes: payment.refunded
# Order Service listens for: payment.refunded
# Order Service marks order as CANCELLED → publishes: order.cancelled
# Notification Service sends: "Sorry, items unavailable — refund initiated"

# Each compensating transaction is also an event in the same Kafka topic
# The full saga history is the append-only event log — perfect audit trail
# No global lock ever held. Each service only locks its own resources.

# Two Kafka producers on different servers, clocks skewed by 80ms

# Server A (clock: 14:23:11.000):
# Produces event: order ORD-001 placed at 14:23:11.000

# Server B (clock: 14:23:10.920, 80ms behind):
# Produces event: order ORD-002 placed at 14:23:10.920

# Both events arrive at the Kafka broker within milliseconds of each other
# But ORD-002's event_time is EARLIER than ORD-001's, even though it was produced LATER
# A consumer sorting by event_time sees ORD-002 before ORD-001 — out of order

# This breaks any logic like:
# "What was the last order before ORD-001?" → might incorrectly return ORD-003
# "What was the total revenue in the 1 second before ORD-001?" → might include ORD-003

# Symptoms in production:
# Windowed aggregations with small windows return slightly wrong counts
# "First order of the day" queries return different results at different times
# Joins on timestamps produce unexpected missing rows

# Solutions:
# 1. Use event time with a watermark (covers clock skew within the watermark window)
#    If watermark = 5 minutes, events up to 5 minutes late are still included correctly
# 2. Use monotonic logical clocks where strict ordering is required
#    (e.g., database transaction IDs, Lamport timestamps)
# 3. Use Kafka's log offset as the ordering key (not event_time)
#    → Guaranteed ordered within a partition, even with clock skew
# 4. NTP synchronisation + monitoring clock drift on all producer machines
#    → Alert if any machine's clock drifts > 50ms

At a data platform team (Flipkart / Meesho):A Spark job is running correctly in staging but producing incorrect aggregation results in production. You investigate and discover the production orders table has severe data skew — one enterprise B2B customer accounts for 30% of all orders. After the shuffle join, one executor processes 30% of the dataset while others process 0.7% each. The straggler completes last, but because it has been running under memory pressure for 4 hours, it has been spilling to disk and producing slightly corrupted partial aggregations. The fix is salting the join key. Understanding shuffle joins and data skew is what lets you find this in 2 hours instead of 2 weeks.

At a fintech during an incident (Razorpay):The payment database's follower replica is 4 minutes behind the leader. A downstream analytics pipeline that reads from the follower is showing payment counts from 4 minutes ago. Someone suggests querying the leader directly. You push back — the leader is handling production write traffic and adding read queries to it risks impacting payment processing latency. Instead you identify why the follower is lagging: a long-running analytical query is blocking WAL replay. You kill the query, the follower catches up in seconds. Understanding leader-follower replication and replication lag is what makes you the person who solves the incident, not the person who makes it worse.

In a system design interview: "Design a distributed pipeline that processes 10 million orders per day and guarantees that each order is counted exactly once in the daily revenue report, even if nodes fail mid-processing." The interviewer is listening for: checkpointing strategy (how far back do you replay on failure), idempotent writes (why an upsert on order_id is needed, not an INSERT), delivery semantics (at-least-once delivery with idempotent consumers = exactly-once behaviour), and partition strategy (what is the partition key, how do you avoid hot partitions). Every one of these is covered in this module and the modules before it.