DDIA Ch 10 — Notes & Questions

Chapter 10: Batch Processing — Study Notes

Batch processing reads a bounded input dataset (known, finite size) and produces a
derived output, without modifying the input. Because input is bounded, a job knows when
it’s done and eventually completes. This chapter traces the lineage of design ideas from
Unix tools → MapReduce → modern dataflow engines, and shows that the same principles
(immutable inputs, composable tools, “do one thing well”) recur at every scale.

Three system types are contrasted throughout: - Services (online): wait for requests,
optimize for low response time. - Batch processing (offline): crunch large bounded inputs,
optimize for throughput. - Stream processing (near-real-time): unbounded inputs, covered
in Ch. 11.

Batch Processing with Unix Tools

Simple Log Analysis

A chain of Unix commands ( awk | sort | uniq -c | sort -r -n | head ) can find the top-
N URLs in an nginx log, processing gigabytes in seconds.
The pipeline is easily tweaked (e.g., filter CSS, count IPs instead) — surprisingly
powerful with awk, sed, grep, sort, uniq, xargs .

Sorting vs. In-Memory Aggregation

Two ways to count occurrences: - In-memory hash table (e.g., a Ruby script): keeps a
counter per distinct key. Working set = memory needed for random access, depends only
on the number of distinct keys. Fine if it fits in RAM. - Sorting (the Unix approach): repeated
keys become adjacent after sorting, then counted sequentially. Wins when the working set
exceeds memory — sort spills to disk and parallelizes across cores, using efficient
sequential I/O (same SSTable/LSM-tree merge-sort principle from Ch. 3). - Rule of thumb:
distinct keys fit in memory → hash aggregation; otherwise → sorting.

The Unix Philosophy

Coined from Doug McIlroy’s 1964 pipe analogy. Key tenets (1978): 1. Make each program
do one thing well; build afresh for new jobs rather than bloating old ones. 2. Expect output
to become another program’s input. Don’t clutter output; avoid columnar/binary/interactive
formats. 3. Build software to be tried early; throw away clumsy parts and rebuild. 4. Use
tools over unskilled labor, even if you must build (and later discard) the tool.

This mirrors today’s Agile/DevOps values — little has changed in four decades.

What makes Unix composable: - Uniform interface: everything is a file (file descriptor) —
an ordered sequence of bytes. Files, pipes, sockets, devices all share it. By convention
many tools treat the bytes as ASCII text with \n -separated records. (Imperfect — {print
$7} is unreadable — but interoperable. Analogous uniform interface: URLs + HTTP on the

web.) - Separation of logic and wiring: programs use stdin/stdout; the shell user wires
inputs/outputs. This is loose coupling / late binding / inversion of control. Limits: multiple
inputs/outputs are awkward, and you can’t pipe directly into a network connection. -
Transparency & experimentation: inputs are immutable (rerun freely), you can inspect
intermediate output ( | less ), and stage outputs to files to restart later stages. - Biggest
limitation: Unix tools run on a single machine — hence Hadoop.

MapReduce and Distributed Filesystems

MapReduce is like distributed Unix tools across thousands of machines: a blunt but
effective brute-force tool. A job takes inputs, produces outputs, normally has no side effects,
and writes output files once, sequentially.

Distributed Filesystems (HDFS)

MapReduce reads/writes a distributed filesystem (HDFS = open-source GFS

reimplementation). Similar: GlusterFS, QFS; object stores like S3, Azure Blob,
OpenStack Swift.
Shared-nothing principle (vs. shared-disk NAS/SAN with Fibre Channel) — only
commodity machines + conventional datacenter network, no special hardware.
A NameNode daemon tracks which file blocks live on which machine; conceptually
one giant filesystem spanning all disks.
Fault tolerance: blocks replicated across machines, or erasure coding (e.g., Reed–
Solomon) for lower storage overhead than full replication. (Erasure coding loses data-
locality advantage.)
 Scales to tens of thousands of machines, hundreds of petabytes, cheaply on
commodity hardware.

MapReduce Job Execution

Four steps, mapping directly onto the Unix log example: 1. Read & split into records (input
format parser; \n = record separator). 2. Mapper — called once per record, extracts key +
value. Stateless, independent per record; emits 0..N key-value pairs. 3. Sort by key —
implicit, you never write it. Mapper output is always sorted before reaching the reducer. 4.
Reducer — called per distinct key with an iterator over all its values; produces output
records.

You only write mapper and reducer. For multi-step logic (e.g., top-N requires a second
sort), chain jobs into workflows.

Distributed execution: - Parallelism via partitioning (Ch. 6). Each input file block = one
map task (often hundreds of MB). - Putting computation near the data: scheduler runs a
mapper on a machine that already holds a replica of the input → reduces network load,
increases locality. Framework copies the code (JAR) to the machine. - # map tasks = #
input blocks; # reduce tasks = configured by job author. - Keys are routed to reducers by
hash of key. Each mapper partitions its output by reducer and writes sorted files locally
(SSTable-style). - The shuffle: the process of partitioning by reducer, sorting, and copying
partitions from mappers to reducers (no randomness — confusing name). - Reducers merge
sorted mapper outputs (preserving sort order) and call the reducer function per key.

MapReduce workflows: - Jobs chained implicitly by directory name (job 1 output dir = job
2 input dir). Less like Unix pipes (small in-memory buffer), more like commands passing
data through temporary files. - A job’s output is valid only on successful completion
(partial output of failed jobs is discarded), so a downstream job can’t start until upstream
jobs succeed. - Workflow schedulers handle dependencies: Oozie, Azkaban, Luigi, Airflow,
Pinball. Recommendation-system workflows commonly chain 50–100 jobs. - Higher-level
tools that auto-wire stages: Pig, Hive, Cascading, Crunch, FlumeJava.

Reduce-Side Joins and Grouping

A join in batch context = resolving all occurrences of an association across the whole
dataset (not a per-user index lookup). MapReduce does full table scans — fine for
aggregates over many records, terrible for fetching a few.

Don’t query a remote DB per record: throughput limited by round-trip time, caches
depend on data distribution, parallel queries can overwhelm the DB, and remote reads
 make the job nondeterministic. Instead, copy the DB into HDFS (e.g., via ETL) so
the join is local and deterministic.

Sort-merge join (reduce-side): - Mappers from both inputs extract the join key (e.g., user
ID). Partitioning + sorting bring all records with the same key adjacent in the reducer input. -
Secondary sort arranges records so the reducer sees, e.g., the user-profile record first,
then activity events. Reducer keeps one record in memory, never makes network requests. -
Mental model: mappers “send messages” to reducers — the key is the destination
address. MapReduce separates network communication from application logic and
transparently retries failed tasks.

GROUP BY: same “bring related data together” pattern — set the grouping key as the
mapper output key. Used for counts, sums, top-k, and sessionization (collating a user’s
events, e.g., for A/B testing).

Handling skew (hot keys / linchpin objects): - A single very-popular key (e.g., a celebrity)
overloads one reducer (a hot spot); the whole job waits for the slowest reducer. - Pig
skewed join: sampling job finds hot keys, then spreads hot-key records across several
random reducers; the other input is replicated to all those reducers. - Crunch sharded
join: similar, but hot keys specified explicitly. - Hive skewed join: hot keys declared in
metadata, stored separately, handled via map-side join. - Two-stage grouping: stage 1
groups hot-key records on random reducers into partial aggregates; stage 2 combines them.

Map-Side Joins

No reducers, no sorting — each mapper reads one input block and writes one output.
Requires assumptions about input layout; faster when applicable.

Broadcast hash join: small dataset fits in memory → each mapper loads it into an in-
memory hash table and scans the large input, looking up each record. “Broadcast” =
small input sent to all partitions; “hash” = hash table. (Names: Pig “replicated join”,
Hive “MapJoin”, also Impala.) Alternative: store small input as a read-only on-disk
index (page cache keeps hot parts fast).
Partitioned hash join: both inputs partitioned the same way (same key, hash, #
partitions) → each mapper handles only its matching partition, smaller hash tables.
(Hive: “bucketed map joins.”) Works only if prior jobs produced this partitioning.
Map-side merge join: inputs partitioned and sorted the same way → mapper merges
incrementally, no need to fit in memory.

Output layout matters downstream: reduce-side join output is partitioned/sorted by join

key; map-side join output is partitioned/sorted like the large input. Optimizing joins requires
knowing physical layout (partition count, partition/sort keys) — tracked in HCatalog / Hive
metastore.

The Output of Batch Workflows

Batch processing is neither OLTP nor analytics, though closer to analytics. Output is often a
structure, not a report: - Search indexes: Google’s original MapReduce use (5–10 jobs).
Mappers partition documents, reducers build per-partition (document-partitioned) indexes;
index files are immutable. Rebuild wholesale, or build incrementally (Lucene-style segment
files). - Key-value / ML output: classifiers, recommendation systems. Build the database
files inside the batch job and write them to HDFS, then bulk-load into read-only servers
(Voldemort, Terrapin, ElephantDB, HBase bulk loading). Voldemort serves old files while
copying new ones, then atomically switches over (easy rollback). - Don’t write to an
external DB record-by-record from within a job: slow (per-record network calls), can
overwhelm the DB, and breaks MapReduce’s all-or-nothing guarantee (externally visible
side effects from partial/speculative tasks).

Philosophy of batch outputs (Unix-like): inputs immutable, output fully replaced, no side
effects → - Human fault tolerance: buggy code? roll back code and rerun, or switch back
to old output dir. (Read-write DBs can’t undo bad writes this way.) Minimizing irreversibility
enables fast/Agile development. - Automatic task retry is safe only because inputs are
immutable and failed-task output is discarded. - Same files reusable across jobs
(incl. monitoring jobs). Logic/wiring separation enables code reuse. - Hadoop reduces Unix’s
text-parsing overhead via structured formats: Avro and Parquet (schema-based encoding,
schema evolution).

Comparing Hadoop to Distributed Databases

Hadoop ≈ distributed Unix (HDFS = filesystem, MapReduce = a quirky process that always
runs sort between map and reduce). The parallel join algorithms existed a decade earlier in
MPP databases (Gamma, Teradata, Tandem NonStop SQL).

Diversity of storage: MPP requires up-front schema modeling; HDFS stores any
byte sequence. Hadoop enables dumping raw data first, interpreting later → schema-
on-read, the “sushi principle” (raw data is better), the data lake / enterprise data
hub. Shifts interpretation burden from producer to consumer; speeds centralized data
collection. Hadoop often used for ETL.
Diversity of processing: MPP is monolithic and SQL-centric (great for BI tools like
Tableau). MapReduce lets you run arbitrary code over large datasets — essential for
ML, search ranking, image analysis. Multiple processing models (SQL, MapReduce,
 others) coexist on one shared cluster over the same files (e.g., HBase OLTP +
Impala MPP-analytics, both on HDFS).
Designing for frequent faults: MPP aborts/restarts the whole query on a node crash
(queries are short) and keeps data in memory. MapReduce retries at task granularity
and eagerly writes to disk — better for long jobs likely to hit a failure.
Why so fault-tolerant? Google runs mixed-use datacenters with priority-based
preemption (low-priority batch jobs “pick up scraps”). A 1-hour task has ~5%
preemption risk (>10× hardware failure rate); a 100-task/10-min job has >50%
chance of losing a task. Task-level recovery enables high resource utilization via
overcommitment.
Open-source schedulers (YARN, Mesos, Kubernetes) use preemption less, so
these MapReduce trade-offs make less sense there.

Beyond MapReduce

MapReduce is a useful learning tool (a clear abstraction), but hard to use directly (you’d
hand-code join algorithms) and slow for some workloads. Higher-level APIs (Pig, Hive,
Cascading, Crunch) ease usage but don’t fix the execution model’s problems.

Materialization of Intermediate State

Materialization = eagerly computing and writing out intermediate results to files

(vs. computing on demand). MapReduce fully materializes intermediate state to HDFS.
Recommendation workflows (50–100 jobs) carry lots of intermediate state. Downsides
vs. Unix pipes:
A job can’t start until all tasks of preceding jobs finish — stragglers slow
everything (pipes start processes concurrently, streaming as produced).
Redundant mappers that just re-read a reducer’s output (could be folded into
the prior reducer).
Intermediate files are replicated across nodes — overkill for temporary data.

Dataflow Engines (Spark, Tez, Flink)

Handle an entire workflow as one job (not independent subjobs); model dataflow explicitly
as a DAG of operators. Operators generalize map/reduce and can be connected flexibly: -
repartition + sort by key (enables sort-merge joins / grouping), - partition without sorting
(partitioned hash joins), - broadcast one operator’s output to all partitions (broadcast hash
joins).
Advantages over MapReduce: - Sorting done only where required, not between every
stage. - No unnecessary map tasks (a mapper that doesn’t repartition folds into the
preceding reduce operator). - Explicit dependencies → scheduler makes locality
optimizations (colocate producer/consumer, exchange via shared memory). - Intermediate
state kept in memory or on local disk (less I/O than HDFS replication). - Operators start
as soon as input is ready (no waiting for the whole prior stage). - Reuse JVM processes
(less startup overhead — MapReduce launches a new JVM per task). - Existing
Pig/Hive/Cascading workflows can switch MapReduce → Tez/Spark via config, no code
changes. - Tez = thin library on YARN shuffle service; Spark/Flink = big frameworks with own
networking, scheduler, APIs.

Fault tolerance (dataflow engines): - They avoid materializing to HDFS, so on a node

failure they recompute lost intermediate state from earlier stages / original input. - The
framework tracks lineage: Spark RDDs (track ancestry), Flink checkpoints operator state.
- Recomputation requires deterministic operators — otherwise downstream operators
must be killed and rerun (to avoid contradictions / cascading faults). Sources of accidental
nondeterminism: hash-table iteration order, random numbers, system clock, external data.
Fix e.g. with a fixed random seed. - Recompute isn’t always cheaper: if intermediate data is
small or computation is CPU-intensive, materialize instead. - A sort must consume all input
before producing output (last record could be smallest key) → sorting operators accumulate
state; other parts can pipeline. Flink emphasizes pipelined execution. Final outputs still
land on HDFS (immutable input, replaced output).

Graphs and Iterative Processing

Batch graph analysis (e.g., PageRank, transitive closure) traverses edges repeatedly
until a condition/convergence is met.
Naming caution: dataflow-engine DAGs structure the flow of data as a graph; graph
processing means the data itself is a graph.
“Repeat until done” can’t be expressed in plain MapReduce (single pass). Naive
iterative approach: external scheduler reruns a batch job each iteration — inefficient
because each pass re-reads the whole dataset even if little changed.

Pregel / Bulk Synchronous Parallel (BSP) model (Apache Giraph, Spark GraphX, Flink
Gelly): - One vertex “sends a message” to another vertex (usually along edges); per
iteration, a function processes each vertex’s incoming messages — like a reducer call. -
Unlike MapReduce, a vertex retains its state in memory across iterations; only new
messages are processed. Idle parts of the graph do no work. - Like the actor model, but
vertex state/messages are durable and fault-tolerant, and communication is in fixed rounds
(all messages from iteration N delivered in N+1). Messages processed exactly once despite
unreliable networks. - Fault tolerance: periodic checkpointing of all vertex state to durable
storage; roll back to last checkpoint on failure (or selectively recover one partition if
deterministic + messages logged). - Parallel execution: “thinking like a vertex” —
framework partitions graph (usually arbitrarily by vertex ID) and routes messages.
Downside: heavy cross-machine communication; intermediate messages often bigger
than the graph. - Rule of thumb: if the graph fits in memory (or on disk) of one machine, a
single-machine algorithm (e.g., GraphChi) often beats a distributed one. Go distributed
(Pregel) only when the graph exceeds one machine.

High-Level APIs and Languages

Physical scaling is “solved”; focus shifted to programming model, efficiency, and breadth. -
High-level APIs (Hive, Pig, Cascading, Crunch; Spark/Flink dataflow APIs, à la FlumeJava)
use relational building blocks (join, group, filter, aggregate) and enable interactive,
incremental development (Unix-like). - Toward declarative queries: declaring joins (not
coding them) lets cost-based query optimizers pick the best join algorithm and reorder
joins to minimize intermediate state (Hive, Spark, Flink). - But MapReduce-lineage systems
differ from pure SQL: built on function callbacks (arbitrary user code) → can use rich
library ecosystems (parsing, NLP, image/statistical analysis) with normal package managers.
This arbitrary-code freedom is their lasting advantage over MPP databases. - Declarative
features also help low-level efficiency: column-oriented storage (read only needed columns),
vectorized execution (tight CPU-cache-friendly loops), code generation (Spark → JVM
bytecode, Impala → LLVM native code). - Specialization for domains: reusable libraries
for ML/statistics (Mahout on MapReduce/Spark/Flink; MADlib inside MPP HAWQ),
spatial/similarity search (k-nearest neighbors), genome analysis (approximate string
matching). - Convergence: batch engines gain declarative operators + optimizers (look
more like MPP DBs); MPP DBs become more programmable. In the end, all are “just
systems for storing and processing data.”

Key Takeaways

Batch jobs process bounded input and produce derived output without mutating
the input — so they’re rerunnable, debuggable, and roll-back-friendly (human fault
tolerance).
Unix design principles — immutable inputs, a uniform interface, composable
single-purpose tools, separation of logic and wiring — scale up directly to
MapReduce and dataflow engines.
The two core problems batch frameworks solve are partitioning (bring related data,
e.g., same-key records, together) and fault tolerance (retry safely because tasks are
stateless with no external side effects).
Know the join algorithms: sort-merge (reduce-side, no input assumptions, but
expensive shuffle); broadcast hash (small input loaded in memory by every mapper);
partitioned hash (both inputs pre-partitioned identically).
Handle skew/hot keys by spreading a hot key across multiple reducers
(sampling/explicit hints) or two-stage aggregation.
Don’t query or write external databases per record from a job — copy data into
HDFS for joins, and build immutable DB files for bulk load on output.
Hadoop vs. MPP: HDFS allows schema-on-read / data-lake dumping of arbitrary data
and arbitrary code processing on a shared cluster; MPP gives tuned SQL
performance but rigid up-front modeling.
MapReduce’s heavy disk materialization and task-level retry suit long jobs in
preemption-heavy clusters (Google’s ~5%/hour preemption rate), trading speed for
robustness.
Dataflow engines (Spark/Tez/Flink) treat a whole workflow as one DAG, skip
unnecessary sorting/materialization, pipeline execution, and recompute lost state via
lineage — needing deterministic operators to avoid cascading recomputation.
Graph/iterative algorithms use the Pregel/BSP model (stateful vertices, message
passing in fixed rounds, checkpointed) — but a single machine often wins unless the
graph is huge.
High-level APIs are converging batch engines and MPP databases by combining
declarative optimization (cost-based join selection, vectorization, columnar reads)
with arbitrary-code extensibility.
Chapter 10: Batch Processing - Assessment
Questions

Section 1: Conceptual/Reasoning Questions

Question 1

Why does the Unix philosophy of small, composable tools translate well to MapReduce?
What are the key parallels, and where does the analogy break down?

Question 2

MapReduce was designed to tolerate frequent task termination not primarily because
hardware is unreliable, but for a different reason. What is that reason, and how does it
influence the design trade-offs MapReduce makes (e.g., eagerness to write to disk, task-
level recovery)?

Question 3

The chapter describes the concept of “human fault tolerance” in the context of batch
processing outputs. Explain what this means and why treating inputs as immutable and
avoiding side effects contributes to it. How does this contrast with writing directly to a read-
write database from a batch job?

Question 4

Explain why making operators deterministic is critical for fault tolerance in dataflow engines
like Spark and Flink. What are three sources of nondeterminism that could “accidentally
creep in,” and what would happen if a failed operator were recomputed nondeterministically?

Question 5
The chapter states that for graph processing, “if your graph can fit in memory on a single
computer, it’s quite likely that a single-machine algorithm will outperform a distributed batch
process.” Explain why this is the case, despite the distributed system having far more
aggregate resources.

Question 6

Why does MapReduce have “no concept of indexes” and instead perform full table scans?
Under what circumstances is this actually a reasonable design choice rather than a
limitation?

Section 2: Scenario-Based Problems

Question 7

You need to join a 1TB user activity log with a 10GB user profiles table to produce per-user
activity summaries. Describe how you would implement this as: - (a) A reduce-side sort-
merge join - (b) A map-side broadcast hash join

Under what conditions is each approach preferable?

Question 8

You are building a recommendation system pipeline that consists of 50 chained MapReduce
jobs. The pipeline takes 8 hours to run end-to-end. Your team wants to migrate to a dataflow
engine (e.g., Spark). Identify at least four specific sources of inefficiency in the MapReduce
approach that the dataflow engine would eliminate, and explain the mechanism by which
each improvement works.

Question 9

Your social network has a “celebrity problem”: 0.01% of users have over 10 million followers
each. You need to run a MapReduce job that joins user posts with follower lists to generate
personalized feeds. Describe: - (a) Why this causes a problem in a standard MapReduce job
- (b) Two different algorithmic approaches to handle this skew (describe the mechanism of
each) - (c) The trade-offs between these approaches
Question 10

You are designing a batch pipeline that builds a full-text search index for 500 million
documents stored in HDFS. The final index must be served by a fleet of query-serving
machines. Describe: - (a) How you would use MapReduce to build the index - (b) How you
would get the completed index files to the serving machines - (c) How you would handle
index updates when only 0.1% of documents change daily

Question 11

A data engineer proposes writing MapReduce output directly to a production PostgreSQL

database from within reducers. List at least four problems with this approach, and describe
the better alternative pattern discussed in the chapter.

Question 12

You have two datasets that need to be joined repeatedly by different downstream jobs.
Dataset A (click events) is 50TB and Dataset B (product catalog) is 500GB. Both are
partitioned by product_id with 1000 partitions and sorted by timestamp within each partition.
What type of map-side join can you use, and why? What must be true about the datasets for
this to work?

Section 3: Mathematical/Quantitative Questions

Question 13

A MapReduce job processes 100TB of input across 1,000 mapper tasks. Each mapper
produces 2GB of intermediate output, which must be shuffled to 500 reducers. Assume the
network bandwidth available per node is 10 Gbps and that each reducer must pull data from
all mappers.

a. What is the total volume of data that must be transferred during the shuffle
phase?
b. Assuming uniform distribution across reducers, how much data does each
reducer receive?
 c. What is the minimum theoretical shuffle time, assuming each reducer can
simultaneously download from all mappers and the bottleneck is the reducer’s
inbound bandwidth?

Question 14

At Google, a MapReduce task running for one hour has approximately a 5% chance of being
preempted. Consider a job with 200 tasks, each running for 30 minutes.

a. What is the probability that any individual task is preempted (assume preemption
probability scales linearly with time)?
b. What is the probability that at least one task in the entire job is preempted?
c. If instead of task-level recovery the entire job had to restart on any failure, and
assuming restarts reset the clock, what is the expected number of complete job
attempts needed?

Question 15

A sort-merge join processes two datasets: Dataset A has 10 billion records (1TB) and
Dataset B has 100 million records (50GB). The join key has 100 million distinct values.
Assuming uniform distribution:

a. On average, how many records from Dataset A share the same join key?
b. If you configure 1,000 reducers, approximately how many distinct keys does
each reducer handle?
c. How much data does each reducer process on average?

Section 4: Compare & Contrast Questions

Question 16

Compare MapReduce with MPP databases (like Teradata) across the following dimensions:
- Fault tolerance strategy - Schema flexibility (schema-on-write vs. schema-on-read) -
Intermediate data handling - Query optimization approach - Suitability for different workload
types
Question 17

Compare the three join strategies discussed in the chapter – sort-merge join (reduce-side),
broadcast hash join (map-side), and partitioned hash join (map-side) – on the following
criteria: - Assumptions required about input data - Network data transfer volume - Memory
requirements - Output partitioning and sorting properties

Question 18

Compare MapReduce’s materialization of intermediate state with the pipelined execution

approach of dataflow engines (Spark/Flink/Tez). Discuss: - How each handles intermediate
data between processing stages - The fault tolerance implications of each approach - The
impact on end-to-end job latency - When you might prefer MapReduce’s approach despite
its inefficiency

Question 19

Compare the Pregel (BSP) model of graph processing with implementing iterative graph
algorithms using chained MapReduce jobs. Address: - How state is maintained between
iterations - Communication patterns - Efficiency for sparse graph updates - Fault tolerance
mechanisms

Section 5: Design Questions

Question 20

Design a batch pipeline that computes friend-of-friend (FoF) recommendations for a social
network with 1 billion users, where the average user has 200 friends. The output should be,
for each user, a ranked list of the top 20 suggested connections (friends of friends who are
not already direct friends).

Address: - (a) The data model and storage format - (b) The algorithm expressed as
MapReduce or dataflow stages - (c) How you handle the “celebrity” problem (users with
millions of friends) - (d) An estimate of the intermediate data volume - (e) Whether you would
use MapReduce, a dataflow engine, or a graph processing framework, and why
Question 21

Design a batch processing system that processes 10TB of daily web server logs to produce:
1. Per-URL page view counts (hourly granularity) 2. Per-user session reconstructions
(sessionization) 3. A join of user activity with a 100GB user demographics database to
compute age-group breakdowns per page

The system should run daily and complete within 2 hours. Describe the workflow of jobs, the
join strategies you would use for each stage, and how you would handle late-arriving data.

Answer Key

Answer 1

Key parallels: - Uniform interface: Unix uses files/pipes (byte streams); MapReduce uses
the distributed filesystem (HDFS). Both allow composability because programs agree on a
data exchange format. - Immutable inputs: Both Unix tools and MapReduce treat inputs as
read-only, enabling re-runs and experimentation. - Separation of logic and wiring: Unix
programs use stdin/stdout without knowing the source/destination; MapReduce jobs
read/write to configured directories without knowing what produces/consumes them. -
Small, focused tools: Unix philosophy says each program does one thing well; MapReduce
jobs are often chained into workflows where each job performs one transformation. -
Transparency/experimentation: Both allow inspecting intermediate outputs and re-running
stages independently.

Where the analogy breaks down: - Materialization: Unix pipes stream data incrementally
with small buffers; MapReduce fully materializes intermediate state to HDFS between jobs,
introducing latency. - No streaming between stages: A MapReduce job must finish entirely
before downstream jobs can start, unlike Unix pipes which process data concurrently. -
Typed vs. untyped: Unix uses untyped ASCII text requiring ad-hoc parsing; Hadoop
supports structured formats (Avro, Parquet) with schema evolution. - Parallelism model:
Unix pipelines are single-machine; MapReduce distributes across thousands of machines
with partitioning, shuffling, and sorting as first-class concepts. - Fault tolerance: Unix has
none (a crashed process kills the pipeline); MapReduce retries failed tasks transparently.
Answer 2

The primary reason is resource preemption in shared clusters. At Google, batch jobs run
at low priority on mixed-use datacenters alongside production services. Higher-priority
processes can terminate batch tasks at any time to reclaim resources – at Google, a task
running for an hour has ~5% chance of preemption, which is an order of magnitude more
frequent than hardware failures.

Design implications: - Eager disk writes: MapReduce writes mapper output to local disk
and reducer output to HDFS. This means if a task is preempted, the work done by
completed tasks is preserved and doesn’t need re-execution. - Task-level recovery: Rather
than restarting entire jobs, only the preempted task needs to re-run. Given that preemption
is frequent, job-level restart would be catastrophically wasteful. - Tolerance of slow
execution: Writing to disk at every stage is slower in the failure-free case, but this overhead
is worthwhile when tasks are frequently killed. - Economic efficiency: This design enables
batch jobs to “pick up the scraps” – using spare resources at low cost, accepting the trade-
off of occasional preemption for much lower resource pricing.

In environments without aggressive preemption (many open-source deployments), these

design choices are less optimal, which is why dataflow engines like Spark can outperform
MapReduce by keeping more data in memory.

Answer 3

“Human fault tolerance” means the ability to recover from bugs introduced by humans –
faulty code that produces incorrect output.

How immutability + no side effects enables this: - If a batch job produces wrong output
due to a code bug, you can fix the code and simply re-run the job. The original input files are
unchanged (immutable), so reprocessing yields correct results. - You can keep the old
(incorrect) output alongside new output and switch back if needed. - Multiple versions of
processing logic can be run against the same input for comparison/validation.

Contrast with writing directly to a database: - If buggy code writes bad data directly to a
production database, rolling back the code does NOT fix the data already written. - You may
not be able to distinguish corrupted records from valid ones. - There’s no clean “undo” –
you’d need backup restoration or complex compensating transactions. - The damage may
propagate to other systems that read from the database before the bug is detected.
The batch philosophy of “inputs in, outputs out” creates a clear audit trail and makes rollback
trivial by simply pointing consumers at the previous output directory.

Answer 4

Why determinism matters: Dataflow engines recover from failures by recomputing lost
intermediate state from available upstream data. If a recomputed operator produces different
output than the original (lost) run, downstream operators that already received the original
output now have inconsistent data – some records were processed against old values, some
against new values. This creates silent data corruption.

The cascade effect: If recomputation produces different results, the framework must also
kill and restart all downstream operators that consumed the original output, potentially
cascading through the entire DAG.

Three sources of accidental nondeterminism: 1. Hash table iteration order: Many

languages (Java HashMap, Python dict pre-3.7) do not guarantee iteration order, so
outputting records by iterating a hash table can produce different orderings across runs. 2.
System clock usage: Using [Link]() or similar in processing logic
produces different values on re-execution. 3. Random number generators: Using
[Link]() or similar without a fixed seed produces different results. Statistical sampling

algorithms often rely on randomness.

Solution: Use fixed seeds for RNGs, avoid clock-dependent logic, sort collections before
iterating, and log/replay external data sources.

Answer 5

Cross-machine communication overhead: Distributed graph algorithms (like Pregel/BSP)

send messages along graph edges. Real-world graphs are poorly partitioned – vertices that
need to communicate frequently end up on different machines. The intermediate state
(messages between vertices) is often larger than the original graph itself.

Network latency vs. memory access: A single-machine algorithm accesses vertex state
via memory pointers (nanoseconds); a distributed algorithm must serialize messages, send
them over the network (milliseconds), and deserialize. This is 5-6 orders of magnitude
slower per operation.
Synchronization barriers: BSP models require all machines to complete an iteration before
the next begins. Stragglers (slow machines) gate the entire computation. Single-machine
algorithms have no such synchronization overhead.

Partitioning difficulty: Finding an optimal graph partition that minimizes cross-partition

edges is itself an NP-hard problem. In practice, graphs are partitioned arbitrarily (by vertex
ID hash), guaranteeing poor locality.

Framework overhead: Distributed systems add overhead for fault tolerance

(checkpointing), coordination, and resource management that a single-machine algorithm
avoids entirely.

The bottom line: the constant factors of distributed execution are so large that they
overwhelm the parallelism benefit unless the graph is genuinely too large for a single
machine.

Answer 6

MapReduce operates without indexes because its design purpose is bulk analytical
processing of entire datasets, not point lookups.

Why full scans are reasonable: - Analytic workloads: When computing aggregates over
large fractions of the data (e.g., all page views, all user sessions), you need to read most
records anyway. An index only helps when you need a small subset. - Schema-on-read
flexibility: Since data can be stored in any format without predefined schema, building
indexes upfront is impractical – you don’t know what queries will be asked. - Parallelism
amortizes cost: A full scan of 100TB distributed across 1,000 machines means each
machine scans only 100GB. With sequential disk reads at ~500MB/s, that’s ~200 seconds
per mapper – entirely feasible. - Write-once, read-many: Data in HDFS is written once and
potentially processed by many different jobs with different access patterns. Building indexes
for all possible access patterns is prohibitively expensive. - Join processing: MapReduce
joins require accessing all records with a given key across multiple datasets simultaneously.
Full scans with sort-merge are more efficient than random lookups when the join involves a
large fraction of records.

The trade-off: this makes MapReduce unsuitable for OLTP-style point queries (use a
database for those), but excellent for batch analytics where you’d touch most of the data
anyway.
Answer 7

(a) Reduce-side sort-merge join: 1. Mapper phase: Two sets of mappers run in parallel.
One set reads the 1TB activity log, extracts (user_id, activity_event) pairs. The other set
reads the 10GB profiles table, extracts (user_id, profile_data) pairs. 2. Shuffle/Sort: The
framework partitions all key-value pairs by user_id (using a hash), and sorts within each
partition. Secondary sort ensures profile records come before activity records for the same
user_id. 3. Reducer: Each reducer receives all records for a set of user_ids, sorted. For
each user_id, it first reads the profile record (storing age, etc. in a local variable), then
iterates over all activity events, emitting enriched activity summaries.

(b) Map-side broadcast hash join: 1. Each mapper loads the entire 10GB user profiles
table into an in-memory hash table (keyed by user_id). 2. Each mapper reads one block of
the 1TB activity log, and for each activity record, looks up the user_id in the hash table to
retrieve the profile. 3. No reducers, no shuffle phase needed.

When each is preferable: - Broadcast hash join is preferable when the small dataset
(10GB) fits in memory on each mapper machine. It avoids the expensive shuffle of 1TB of
data. Modern machines with 64-128GB RAM can easily hold 10GB. - Reduce-side join is
preferable when both datasets are large (neither fits in memory), when you need sorted
output, or when the “small” dataset is actually too large for memory (e.g., 100GB+ profiles
table).

In this specific scenario, the broadcast hash join is clearly superior – it eliminates the
network shuffle of 1TB of data entirely.

Answer 8

Four sources of inefficiency eliminated by dataflow engines:

1. Materialization of intermediate state to HDFS: Between every two MapReduce jobs,

output is written to HDFS (replicated 3x to disk on multiple machines). Dataflow
engines keep intermediate data in memory or on local disk without replication.
Mechanism: Operators pass data directly through in-memory buffers or local spill files,
skipping HDFS serialization/replication.

2. Unnecessary map stages: In chained MapReduce, the mapper of job N+1 often just
reads what reducer N wrote and re-partitions it. Dataflow engines chain operators
directly – a reduce operator’s output feeds into the next operator without an intervening
 map. Mechanism: The DAG scheduler connects operator outputs directly to
downstream operator inputs.

3. Startup overhead: MapReduce launches a new JVM for every map and reduce task.
With 50 jobs, each having hundreds of tasks, that’s tens of thousands of JVM startups.
Dataflow engines reuse JVM processes across operators. Mechanism: Long-lived
executor processes receive new tasks without cold-start overhead.

4. Sequential stage execution: A downstream MapReduce job cannot start until ALL
tasks of the upstream job complete (must wait for stragglers). Dataflow engines can
start downstream operators as soon as their specific input partition is ready (pipelined
execution). Mechanism: The scheduler tracks fine-grained data readiness and starts
operators incrementally.

Additional benefit: The query optimizer in dataflow engines can also reorder joins to
minimize intermediate data volume, and skip sorting where it’s not needed (e.g., for hash
joins).

Answer 9

(a) The problem: In a standard MapReduce job, all records with the same key go to the
same reducer. A celebrity with 10 million followers means 10 million+ records sent to one
reducer. That single reducer becomes a bottleneck – it takes orders of magnitude longer
than other reducers. Since the job completes only when ALL reducers finish, the entire job is
gated by these “hot” reducers (skew/hot spots).

(b) Two approaches:

Approach 1 – Skewed join with random distribution (Pig’s approach): - Run a sampling
job first to identify hot keys. - For hot keys: mappers send records to one of N randomly
chosen reducers (not deterministically by hash). The other side of the join (e.g., celebrity’s
profile) is replicated to ALL N reducers handling that key. - For non-hot keys: standard hash
partitioning. - Result: work for hot keys is parallelized across N reducers instead of one.

Approach 2 – Two-stage aggregation: - First MapReduce stage: append a random

number (0 to N-1) to each hot key, creating N sub-keys. Records distribute uniformly across
N reducers. Each reducer performs a partial aggregation. - Second MapReduce stage: strip
the random suffix and combine partial aggregates into the final result.
(c) Trade-offs: - Approach 1 requires replicating the smaller join input to multiple reducers,
consuming more network bandwidth and memory. It works well when the replicated side is
small. - Approach 2 requires two MapReduce stages (more latency, more materialization),
but doesn’t require replication of join inputs. It works best for aggregation operations (SUM,
COUNT) that are associative and commutative. - Both require knowing which keys are hot
(either via sampling or explicit specification). Pig’s sampling approach is automatic; Hive’s
requires manual metadata annotation.

Answer 10

(a) Building the index with MapReduce: - Mappers: Parse documents, tokenize text,
extract (term, document_id) pairs. Each mapper processes a subset of the 500M
documents. - Partitioning: Partition by term (or by document_id for document-partitioned
indexes). For document-partitioned indexes, partition documents across reducers. -
Reducers: Each reducer builds an inverted index for its partition – creating a term dictionary
mapping terms to postings lists (sorted lists of document IDs containing that term). The
reducer writes the index as immutable files (e.g., Lucene segment files) to HDFS.

(b) Getting index files to serving machines: - Build the index files as output in HDFS (one
set of files per partition/shard). - Copy the completed index files from HDFS to local disk on
serving machines (bulk transfer). - Serving machines continue serving from old index files
during the copy. - Once copy is complete, atomically switch to the new index files (e.g.,
rename/symlink swap). - Keep old files available for rollback if the new index has problems.

(c) Handling daily updates (0.1% change = 500K documents): - Option A (simple): Re-
run the entire indexing pipeline daily. Computationally expensive (reprocessing 500M
documents to update 500K), but simple to reason about – “documents in, indexes out.” -
Option B (incremental): Use Lucene’s segment-based approach – write new/modified
documents as new index segments, mark deleted documents in a deletion bitmap, and
periodically merge segments in the background. This is more complex but far more efficient
for small update ratios. - The chapter suggests that for batch processing, Option A’s
simplicity is valuable despite the compute cost, while Option B (incremental processing)
transitions toward stream processing concepts discussed in Chapter 11.

Answer 11
Four problems with writing directly to a production database from MapReduce:

1. Performance: Making a network request for every record is orders of magnitude

slower than normal batch throughput. Even with client-side batching, it’s far slower
than sequential file writes.

2. Overwhelming the database: MapReduce runs many tasks in parallel. Hundreds of

reducers simultaneously writing at batch-process rates can overwhelm the database,
degrading query performance for other users and causing operational incidents.

3. Loss of all-or-nothing guarantee: MapReduce normally ensures output reflects

exactly-once execution of all tasks. Writing to an external database creates visible side
effects that break this guarantee. Partially completed jobs leave partial (and possibly
duplicate) data visible to other systems due to retries and speculative execution.

4. Non-determinism/external coupling: The batch job becomes coupled to the

database’s availability. If the database is down or slow, the batch job fails or slows. The
job also becomes nondeterministic if the database state changes during execution.

Better alternative: Build the output database files directly within the MapReduce job (as
immutable key-value store files). Write these files to HDFS. Then bulk-load them to serving
nodes: - Serving nodes continue serving old data during the copy. - Atomically switch to new
files once copy completes. - Easy rollback by switching back to old files. - Examples:
Voldemort, Terrapin, ElephantDB, HBase bulk loading.

Answer 12

You can use a map-side merge join because both datasets satisfy the required conditions:
1. Both are partitioned the same way (by product_id, 1000 partitions each) 2. Both use the
same partitioning key (product_id) 3. Both are sorted by the same key within each
partition

How it works: Each mapper reads one partition from Dataset A and the corresponding
partition from Dataset B. Since both are sorted by product_id, the mapper performs a merge
(like merge-sort’s merge step) – reading both files incrementally in ascending key order and
matching records with the same product_id. No hash table is needed, so memory
requirements are minimal regardless of partition size.

Requirements that must be true: - Same number of partitions (1000 each) - Same hash
function for partitioning - Same sort key within partitions - These datasets were likely
produced by prior MapReduce jobs that established this layout

Advantage over broadcast hash join: The 500GB product catalog is too large to fit in
memory on each mapper. The merge join handles this because it only needs a small read
buffer for each file, not the entire dataset in memory.

Answer 13

(a) Total shuffle volume: - 1,000 mappers x 2GB each = 2,000 GB (2 TB) total
intermediate data transferred.

(b) Data per reducer: - 2,000 GB / 500 reducers = 4 GB per reducer.

(c) Minimum theoretical shuffle time: - Each reducer must download 4 GB of data. - At 10
Gbps = 1.25 GB/s inbound bandwidth per reducer. - Time = 4 GB / 1.25 GB/s = 3.2
seconds.

However, this assumes the mapper-side outbound bandwidth is not a bottleneck. Each
mapper sends 2GB total, split across 500 reducers. At 10 Gbps outbound per mapper,
sending 2GB takes 2/1.25 = 1.6 seconds. Since 1.6s < 3.2s, the reducer inbound bandwidth
is indeed the bottleneck.

Minimum shuffle time: ~3.2 seconds (in the ideal case with perfect parallelism and no
overhead). In practice, shuffle times are much longer due to disk I/O, connection setup
overhead, TCP congestion, and the fact that not all mappers finish simultaneously.

Answer 14

(a) Individual task preemption probability: - 5% for a 60-minute task. - Assuming linear
scaling: a 30-minute task has 2.5% probability of preemption.

(b) Probability that at least one task is preempted: - P(no task preempted) = (1 -
0.025)^200 = (0.975)^200 - (0.975)^200 = e^(200 * ln(0.975)) = e^(200 * (-0.02532)) =
e^(-5.064) = 0.0063 - P(at least one preempted) = 1 - 0.0063 = 99.37%

So it is virtually certain that at least one task will be preempted.

(c) Expected number of complete job attempts (without task-level recovery): - P(job
succeeds in one attempt) = (0.975)^200 = 0.0063 - This is a geometric distribution:
E[attempts] = 1/0.0063 = ~159 attempts.

This illustrates why task-level recovery is essential – without it, the job would likely never
complete in any reasonable timeframe. With task-level recovery, only the preempted tasks
(~5 tasks on average out of 200) need to be re-run.

Answer 15

(a) Records per join key in Dataset A: - 10 billion records / 100 million distinct keys = 100
records per key on average.

(b) Distinct keys per reducer: - 100 million distinct keys / 1,000 reducers = 100,000
distinct keys per reducer.

(c) Data per reducer: - From Dataset A: 1 TB / 1,000 reducers = 1 GB per reducer from
the activity data. - From Dataset B: 50 GB / 1,000 reducers = 50 MB per reducer from the
profile data. - Total per reducer: approximately 1.05 GB per reducer.

Note: The reducer memory requirement is modest – for a sort-merge join with secondary
sort, it only needs to hold one profile record in memory at a time while iterating over the 100
activity records for that key.

Answer 16

Dimension MapReduce/Hadoop MPP Databases (Teradata)

Task-level recovery: if a
mapper/reducer fails,
Query-level recovery: if a node fails, the
only that task is re-
Fault entire query is aborted and restarted.
executed using
tolerance Acceptable because queries typically run
immutable input from
seconds to minutes.
HDFS. Designed for
frequent preemption.
 Dimension MapReduce/Hadoop MPP Databases (Teradata)

Schema-on-read: data
can be stored in any Schema-on-write: data must be imported
format (raw text, JSON, into the database’s storage format with a
Schema
Avro, Parquet). Schema predefined schema before it can be
flexibility
interpretation happens at queried. Ensures data quality but slows
processing time. Enables ingestion.
“data lake” approach.

Fully materialized to
HDFS between jobs Kept in memory as much as possible
Intermediate (replicated, written to (e.g., hash tables for joins). Pipelined
data disk). Enables debugging between query operators. Faster but
and reuse but is volatile.
expensive.

No query optimizer – the

programmer explicitly Sophisticated cost-based query
chooses join algorithms optimizers choose join order, join
Optimization
and data flow. Higher- algorithms, and execution plans
level tools (Hive, Pig) add automatically based on statistics.
some optimization.

General-purpose: ML
Specialized for SQL analytics: business
pipelines, ETL, search
Workload intelligence, reporting, ad-hoc queries.
index building, graph
suitability User-defined functions possible but
processing, custom code.
cumbersome.
Runs arbitrary programs.

Key insight: The two are converging – dataflow engines add query optimizers while MPP
databases add extensibility. But they come from different philosophies: Hadoop is a
“general-purpose OS for data” while MPP databases are optimized query engines.

Answer 17
 Sort-Merge Join Broadcast Hash Partitioned Hash Join
Criterion
(Reduce-side) Join (Map-side) (Map-side)

None – works with Both inputs must be

One input must be
any input data partitioned identically
Input small enough to fit
regardless of size, (same key, same hash
assumptions entirely in memory
partitioning, or function, same number
on each mapper.
sorting. of partitions).

Moderate – the
small input is
High – ALL data Low – each mapper
“broadcast” (copied
from both inputs reads only its own
Network to all mappers). The
must be shuffled partition from each
transfer large input is read
across the network input. No shuffle
locally per file block.
to reducers. needed.
No shuffle of the
large input.

Low – reducer
processes records
High – entire small Moderate – only one
sequentially after
input must fit in partition of the small
Memory sort. Only needs to
memory as a hash input must fit in
requirements hold one record
table on each memory per mapper
from the smaller
mapper. (1/N of total).
side in memory at a
time.

Partitioned and Partitioned and

sorted by the join sorted the same way Partitioned by join key
Output
key (useful for as the large input but not necessarily
properties
downstream sort- (by file blocks). NOT sorted by it.
merge operations). sorted by join key.

Answer 18

Intermediate data handling: - MapReduce: Every intermediate result between jobs is

written to HDFS – serialized, replicated (typically 3x), and persisted to disk across multiple
machines. The next job reads it from HDFS. - Dataflow engines: Intermediate data flows
between operators via in-memory buffers, local disk spill (if needed), or network transfer. No
HDFS replication of intermediate state. Pipelined execution means data can be consumed
as it’s produced.

Fault tolerance implications: - MapReduce: Simple recovery – failed tasks re-read their
immutable HDFS input. No upstream re-execution needed. The materialized intermediate
state acts as a checkpoint. - Dataflow engines: If intermediate state is lost (machine failure),
it must be recomputed from upstream data (possibly all the way back to the original HDFS
input). Spark tracks lineage (RDDs); Flink uses periodic checkpoints. Recovery may
cascade upstream.

End-to-end latency impact: - MapReduce: Each job must fully complete before the next
starts. A 50-job workflow has 50 sequential barriers. Stragglers at each stage compound the
delay. HDFS writes add I/O overhead at every stage. - Dataflow engines: Operators can be
pipelined – downstream operators start as soon as their input is available. Only operations
requiring full input (sorting) create barriers. Typically 3-10x faster for multi-stage workflows.

When MapReduce’s approach is preferable: - When intermediate data is useful for

debugging (inspect output of each stage independently). - When intermediate datasets are
consumed by multiple downstream jobs (shared reference datasets). - When the
environment has very high task failure/preemption rates (Google’s mixed-use clusters at
~5%/hour). - When recomputation would be extremely expensive (CPU-intensive
transformations where re-execution costs more than disk I/O for materialization).

Answer 19

Aspect Pregel (BSP) Chained MapReduce

Vertices maintain in-

memory state across
No persistent state – every iteration re-
iterations. Only new
reads the ENTIRE graph from HDFS
State between messages need
and produces a completely new output,
iterations processing. Efficient for
even if only a tiny fraction of vertices
convergence algorithms
changed.
where most vertices
stabilize early.

Communication Vertices send messages All data flows through the full map-sort-
along edges to other shuffle-reduce pipeline every iteration.
 Aspect Pregel (BSP) Chained MapReduce
vertices. Messages are No concept of targeted vertex-to-vertex
batched and delivered at messaging.
iteration boundaries.
Communication is
targeted and sparse.

Excellent – if a vertex
Poor – every iteration processes all
receives no messages, it
vertices regardless of whether they
Efficiency for does no work. As
changed. Reading/writing entire graph
sparse updates algorithms converge,
to HDFS each time is wasteful when
most vertices become
only 0.1% of vertices are active.
inactive.

Periodic checkpoints of
all vertex state. Recovery
options: (1) roll back Automatic – each iteration’s output is
entire computation to last fully materialized on HDFS. A failed
Fault tolerance checkpoint, or (2) task in any iteration simply re-reads its
selectively recover lost HDFS input. Very robust but very
partitions by replaying expensive.
logged messages (if
deterministic).

Summary: Pregel is far more efficient for iterative graph algorithms because it maintains
state and only processes active vertices. MapReduce’s stateless, full-scan model makes it
fundamentally ill-suited for iterative computation, as it cannot exploit the sparsity of updates
between iterations.

Answer 20

(a) Data model and storage: - Friendship graph stored as an adjacency list in HDFS:
(user_id, [friend_id_1, friend_id_2, ...]) - With 1 billion users and average 200 friends:
~200 billion edges total. - Storage: ~200B per edge (two 8-byte IDs + overhead) = ~40TB for
the friendship graph. - Existing friendships also stored as a set per user for fast “already
friends?” lookups.

(b) Algorithm as dataflow stages:

Stage 1: Enumerate FoF pairs - For each user U with friends [F1, F2, …], emit for each
friend Fi: (Fi, U) – meaning “U is a friend of Fi.” - The reducer for Fi receives all of Fi’s
friends, say [A, B, C, U, …]. - For each friend Fi, emit all pairs: (U, A), (U, B), (U, C), …
meaning “A, B, C are friends-of-friends of U via Fi.” - Output: (user_id, candidate_fof_id,
mutual_friend_id)

Stage 2: Count mutual friends and filter - Group by (user_id, candidate_fof_id). Count the
number of mutual friends. - Filter out pairs where candidate is already a direct friend of the
user. - Output: (user_id, candidate_fof_id, mutual_friend_count)

Stage 3: Rank and select top 20 - Group by user_id. Sort candidates by

mutual_friend_count descending. Take top 20.

(c) Celebrity problem: - A user with 10 million friends generates 10M x (10M-1) / 2 potential
FoF pairs in Stage 1 – completely infeasible. - Solution: Cap the FoF enumeration for
celebrities. Only consider a random sample of a celebrity’s friends when generating FoF
pairs. Alternatively, pre-identify celebrities and exclude them from the intermediate FoF
expansion (they can be recommended separately using other signals). - Use the skewed
join technique: for hot keys (celebrities), distribute their friend lists across multiple reducers
randomly and replicate the smaller join input.

(d) Intermediate data volume estimate: - Stage 1 output: each user has ~200 friends,
each friend has ~200 friends. So each user generates up to 200 x 200 = 40,000 FoF pairs. -
Total: 1 billion users x 40,000 pairs x ~20 bytes per pair = ~800 TB of intermediate data
(before celebrity capping). - With celebrity capping (limiting friends enumerated to e.g., 1000
for high-degree nodes), this drops significantly.

(e) Framework choice: - A dataflow engine (Spark) is preferable over raw MapReduce
because: - Multiple stages benefit from pipelined execution and in-memory intermediate
state. - Spark’s built-in graph library (GraphX) provides optimized primitives. - The 800TB
intermediate data volume would be extremely expensive to materialize to HDFS between
stages. - Not a pure graph framework (Pregel) because: the algorithm is essentially two
large joins and a top-k aggregation – these are more naturally expressed as relational
operations than vertex-centric message passing.

Answer 21

Workflow Design (3 parallel paths + 1 dependent join):

Job 1: Per-URL hourly page views - Single MapReduce/Spark stage. - Mapper: parse log
lines, extract (url + hour_bucket) as key, emit count 1. - Reducer: SUM counts per (url,
hour). - Simple aggregation, no joins needed. Completes independently.

Job 2: Sessionization - Mapper: extract (user_id, timestamp, event_data) from logs. -

Partition by user_id, sort by timestamp (secondary sort). - Reducer: iterate through each
user’s events in time order. Split into sessions using a 30-minute inactivity threshold
(configurable). Assign session IDs. - Output: (user_id, session_id, [ordered events]).

Job 3: Age-group breakdown (depends on Job 2’s output being available) - Join
strategy: Use a broadcast hash join since the user demographics database is 100GB.
With modern machines (256GB+ RAM), this fits in memory. If memory is constrained, use a
partitioned hash join – pre-partition both the sessionized activity data and demographics
by user_id. - Mapper: loads demographics into hash table, reads sessionized activity data,
looks up user’s age for each event, emits (url, age_group, page_view_count). - Second
stage: group by (url, age_group) and aggregate.

Join strategies chosen: - Job 3 join: broadcast hash join (100GB demographics fits in
memory on modern hardware) or partitioned hash join (if memory is limited). NOT reduce-
side join, because we want to avoid shuffling 10TB of log data unnecessarily.

Handling late-arriving data: - Run the pipeline on the current day’s data as scheduled
(T+0). - Maintain a “late arrivals” buffer: logs that arrive after the processing window are
accumulated. - Run a daily “catch-up” job that processes only the late-arriving data and
produces incremental updates (delta files). - Merge delta results with the main output (e.g.,
add late page views to the hourly counts, append events to sessions). - Define an
SLA/cutoff: data arriving more than 48 hours late is discarded or processed in a separate
reconciliation batch.

Meeting the 2-hour SLA: - Run Jobs 1 and 2 in parallel (both read from the same 10TB
input). - Job 3 starts after Job 2 completes (needs sessionized output). - With a 100-node
cluster, 10TB / 100 nodes = 100GB per node. At ~500MB/s sequential read, each node
processes input in ~200 seconds. With shuffle and reduce overhead, Jobs 1 and 2 should
complete within ~30-45 minutes. - Job 3 (join + aggregation) adds another ~30 minutes. -
Total: well within the 2-hour budget using a dataflow engine. MapReduce with materialization
between stages would be tighter but still feasible.