Review and interview preparation

"Make it faster" broke it. The checks noticed.

A project review, and a project story

Words you'll meet in this lesson

Architecture at a glance: what the review changed

From our lab (the review experiments)

The review at three levels

The review checklist, with our evidence

Ways you will be asked about this project

Animated example: a fast job that was wrong

Reviewing, fixing and presenting the capstone

1. Find where the time goes

   Launched map tasks=104
   Total time spent by all map tasks (ms)=2049842

   2,049,842 ms / 104 maps ≈ 19.7 s per map for ~3,400 lines each
   The work per map is tiny; the cost is starting a container, a JVM and Python 104 times on 6 vcores (Lesson 7.5's small files problem, now measured).

2. Try combining small files

   -D mapreduce.input.fileinputformat.split.maxsize=8388608 \
   -inputformat org.apache.hadoop.mapred.lib.CombineTextInputFormat

   number of splits:3 real 0m38.006s CLEAN=192799 DUPLICATE=164120 LATE=192799 MALFORMED=400
   7× faster and badly wrong. Never judge an optimisation by its run time; judge it by its counters.

3. Fix the key, find the second bug

   -D stream.map.input.ignoreKey=true \

   real 0m43.720s CLEAN=354135 DUPLICATE=2784 LATE=46646 MALFORMED=296
   Lines are right now, but LATE uses the file day from mapreduce_map_input_file, which is exported once per task. A combined split holds about 35 files.

4. Put the lineage into the data

   .withColumn("fday", F.regexp_extract(F.input_file_name(), r"_(\d{4}-\d{2}-\d{2})\.csv$", 1))
   …partitionBy("college").text(f"{B}/bronze_monthly/orders")

   lines: 357228 8607649 /capstone/bronze_monthly/orders/college=KHR/part-00002-….c000.txt 4192061 /capstone/bronze_monthly/orders/college=NRL/part-00000-….c000.txt 5728542 /capstone/bronze_monthly/orders/college=PNL/part-00001-….c000.txt 7329058 /capstone/bronze_monthly/orders/college=VSH/part-00000-….c000.txt
   Spark's input_file_name() is evaluated per row, so each line gets its own file day. Headers are dropped: 357,332 − 104 = 357,228 lines. Bronze itself stays untouched.

5. Run v2 and prove it equal

   hdfs dfs -cat '/capstone/stage/clean/part-*'    | sort | md5sum
   hdfs dfs -cat '/capstone/stage/clean_v2/part-*' | sort | md5sum

   number of splits:4 LATE=2100 e4cb870499eba837b494c5e4e85582dc - e4cb870499eba837b494c5e4e85582dc - real 0m52.718s
   The mapper change is two lines (read the tag if present). Same counters, same checksum: this is a regression test you can show a reviewer.

6. Failure drill: break worker2 mid-job

   docker exec lab8-worker2-1 bash -c 'pkill -9 -f "[Y]arnChild"; hdfs --daemon stop datanode'

   Task Id : attempt_1792730826465_0014_m_000033_0, Status : FAILED Killed by external signal Task Id : attempt_1792730826465_0014_r_000000_0, Status : FAILED ERROR StreamJob - Error Launching job : java.io.IOException: Unknown Job job_1792730826465_0014 Streaming Command Failed! $ yarn application -status application_1792730826465_0014 Final-State : SUCCEEDED
   The data survived (output md5 identical), but the client "failed". The AM's history file was being written through all 3 DataNodes; with one dead there was no replacement, so the history server never got the job.

7. Fix and repeat the drill

   -D dfs.client.block.write.replace-datanode-on-failure.policy=NEVER \

   Job job_1792730826465_0015 completed successfully Failed map tasks=4 Launched map tasks=108 CLEAN=354135 LATE=2100 e4cb870499eba837b494c5e4e85582dc - real 5m48.109s
   Four killed attempts re-ran elsewhere (108 launched for 104 maps); counters of failed attempts were discarded, so totals stayed exact. Recovery cost about 70 extra seconds.

8. Write it up for interviews

   Built a Hadoop/Spark batch pipeline for 4 canteens (130 daily files, 3.57 lakh records):
   dedupe + quality rules in MapReduce, Parquet silver/gold in Spark, UPI reconciliation;
   10/10 acceptance checks; cut clean time 4m39s → 53s via compaction (byte-identical output).

   Three lines, all measured. The worked examples and the STAR scenario below turn this into a two-minute answer.

The failure drill: the log and the cluster

Our second drill (with the fix), second by second. Left: the client's terminal. Right: what YARN and HDFS do on the three workers.

Reviews and interviews at work

Interview question bank: where each answer lives

The questions that come up again and again in Hadoop and data-engineering interviews, with the lessons that answer them. Revise the linked lesson's "Quick revision" card and its interview questions the night before.

Simplified; organisations differ in process. Kadak Canteens and its data are fictional.

Sources (checked 23 Oct 2026): Apache Hadoop: Hadoop Streaming (keys discarded with TextInputFormat; configuration copied to the environment) · Cloudera Community: replace-datanode-on-failure.policy on a 3-DataNode cluster (quotes hdfs-default.xml)

Why one file name for many files breaks "late"

A simulation of our second combine attempt. One Python process reads lines from three files but is told only one file name at start-up, as Streaming does with its environment.

files = {"orders_KHR_2026-09-12.csv": ["KHR-0912-1,2026-09-12", "KHR-0912-2,2026-09-12"],
         "orders_KHR_2026-09-14.csv": ["KHR-0912-9,2026-09-12", "KHR-0914-1,2026-09-14"],
         "orders_KHR_2026-09-15.csv": ["KHR-0915-1,2026-09-15"]}

def late_count(lines_with_day):
    return sum(1 for fday, line in lines_with_day if fday > line.split(",")[1])

# baseline: one process per file, each knows its own file name
per_file = [(name[-14:-4], line) for name, lines in files.items() for line in lines]
# combined split: one process, one file name for the whole split (here: the last file)
told = list(files)[-1][-14:-4]
combined = [(told, line) for lines in files.values() for line in lines]
# fix: the file day travels inside every line (compaction)
tagged = [(name[-14:-4], line) for name, lines in files.items() for line in lines]

print("one file per map :", late_count(per_file))
print("combined, env var:", late_count(combined))
print("lineage in data  :", late_count(tagged))

• told = list(files)[-1][-14:-4]Like mapreduce_map_input_file: fixed when the process starts, wrong for the other files in the split.
• tagged = [(name[-14:-4], line) for name, lines in files.items() for line in lines]compact.py does this with input_file_name(): each line carries its own file day.

Run it on your PC: the review experiments

Same container and folder as Lessons 14.1–14.2, after the baseline clean job. Job logs are filtered to the lines that matter (splits, attempts, counters, timing).

# combine_job.sh: the same clean job, but pack the 104 small files into a few splits
hdfs dfs -rm -r -f -skipTrash /capstone/stage/clean_combined
mapred streaming \
  -D mapreduce.job.name=capstone-clean-combined \
  -D mapreduce.job.reduces=3 \
  -D mapreduce.input.fileinputformat.split.maxsize=8388608 \
  -files clean_mapper.py,clean_reducer.py,capstone_raw/menu.csv \
  -inputformat org.apache.hadoop.mapred.lib.CombineTextInputFormat \
  -input '/capstone/bronze/orders/college=*' \
  -output /capstone/stage/clean_combined \
  -mapper clean_mapper.py -reducer clean_reducer.py

# combine_job.sh: the same clean job, but pack the 104 small files into a few splits
hdfs dfs -rm -r -f -skipTrash /capstone/stage/clean_combined
mapred streaming \
  -D mapreduce.job.name=capstone-clean-combined \
  -D mapreduce.job.reduces=3 \
  -D mapreduce.input.fileinputformat.split.maxsize=8388608 \
  -D stream.map.input.ignoreKey=true \
  -files clean_mapper.py,clean_reducer.py,capstone_raw/menu.csv \
  -inputformat org.apache.hadoop.mapred.lib.CombineTextInputFormat \
  -input '/capstone/bronze/orders/college=*' \
  -output /capstone/stage/clean_combined \
  -mapper clean_mapper.py -reducer clean_reducer.py

# compact.py: 104 small daily files -> one file per canteen, each line tagged with the day it arrived
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("capstone-compact").getOrCreate()
B = "hdfs://namenode/capstone"
lines = (spark.read.text(f"{B}/bronze/orders/college=*")
         .withColumn("fday", F.regexp_extract(F.input_file_name(), r"_(\d{4}-\d{2}-\d{2})\.csv$", 1))
         .withColumn("college", F.regexp_extract(F.input_file_name(), r"college=(\w+)", 1))
         .where(~F.col("value").startswith("order_id,")))
(lines.select(F.concat_ws("\t", "fday", "value").alias("value"), "college")
      .repartition("college").write.mode("overwrite").partitionBy("college").text(f"{B}/bronze_monthly/orders"))
print("lines:", spark.read.text(f"{B}/bronze_monthly/orders").count())

export SPARK_HOME=/tmp/opt/spark-4.0.4-bin-hadoop3 HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
$SPARK_HOME/bin/spark-submit --master yarn --num-executors 3 --executor-memory 768m --driver-memory 768m compact.py
hdfs dfs -ls -R /capstone/bronze_monthly/orders | grep part- | awk '{print $5, $8}'

# clean_mapper_v2.py = clean_mapper.py plus two lines after line.rstrip():
#     if "\t" in line:
#         fday, line = line.split("\t", 1)
# clean_job_v2.sh: the clean job on 4 compacted files (one per canteen)
hdfs dfs -rm -r -f -skipTrash /capstone/stage/clean_v2
mapred streaming \
  -D mapreduce.job.name=capstone-clean-v2 \
  -D mapreduce.job.reduces=3 \
  -files clean_mapper_v2.py,clean_reducer.py,capstone_raw/menu.csv \
  -input '/capstone/bronze_monthly/orders/college=*' \
  -output /capstone/stage/clean_v2 \
  -mapper clean_mapper_v2.py -reducer clean_reducer.py
hdfs dfs -cat '/capstone/stage/clean/part-*' | sort | md5sum
hdfs dfs -cat '/capstone/stage/clean_v2/part-*' | sort | md5sum

# drill.sh: run the clean job and break worker2 while it runs (run from the host, not a container)
docker exec -w /tmp/capstone lab8-namenode-1 bash clean_job.sh > tx_drill.txt 2>&1 &
sleep 75
docker exec lab8-worker2-1 bash -c 'pkill -9 -f "[Y]arnChild"; hdfs --daemon stop datanode'   # kill running tasks + DataNode
wait
docker exec lab8-worker2-1 hdfs --daemon start datanode

# afterwards, inside the namenode container:
# yarn application -status application_<id> | grep State
# yarn logs -applicationId application_<id> -log_files syslog | grep JobHistoryEventHandler

# clean_job_small.sh: clean_job.sh for a cluster of only 3 DataNodes (keep writing if one dies)
hdfs dfs -rm -r -f -skipTrash /capstone/stage/clean
mapred streaming \
  -D mapreduce.job.name=capstone-clean \
  -D mapreduce.job.reduces=3 \
  -D dfs.client.block.write.replace-datanode-on-failure.policy=NEVER \
  -files clean_mapper.py,clean_reducer.py,capstone_raw/menu.csv \
  -input '/capstone/bronze/orders/college=*' \
  -output /capstone/stage/clean \
  -mapper clean_mapper.py -reducer clean_reducer.py

A regression check, line by line

The check that rejected the second combine attempt: compare every counter of a new run with the trusted baseline and list the differences.

Worked examples: real problems, solved and traced in memory

Three calculations a reviewer or interviewer will ask you to do on the spot.

2,049,842 ms · 104 maps · ~1 s real work per map

avg seconds per map, overhead %: (19.7, 95)

12.4 MB per canteen-month · 40 canteens · 36 months · rf 3

raw GB, with replicas GB: (17.5, 52.4)
order files per year: 12480

4 attempts × ~20 s · 348 s vs 279 s

extra seconds, not explained by re-runs: (69, 49)

Check yourself

Quick revision

Interview and viva questions

Practice problems

Review calculations and interview drills using the experiment numbers.

279 s · 53 s · 77 s

5.3 x without compaction, 2.1 x including it

print(round(279 / 53, 1), "x without compaction,", round(279 / (53 + 77), 1), "x including it")

two layouts

3436 89307

print(round(357332 / 104), round(357228 / 4))

108 · 4

104

print(108 - 4)

192,799 vs 354,135 · 164,120 vs 2,784

CLEAN -161336 -46%
DUPLICATE 161336 +5795%

for name, got, want in [("CLEAN", 192799, 354135), ("DUPLICATE", 164120, 2784)]:
    print(name, got - want, f"{(got - want) / want:+.0%}")

4 sizes

6.5 MB, 4.8% of a block

sizes = [8607649, 4192061, 5728542, 7329058]
avg = sum(sizes) / len(sizes)
print(round(avg / 1e6, 1), "MB,", f"{avg / (128 * 1024 * 1024):.1%} of a block")

your project numbers

10/10 checks; found Rs 43,710 undercharged and a Rs 6,854 UPI gap; clean job 279s -> 53s (5.3x)

r = {"checks": "10/10", "undercharged": 43710, "upi_gap": 6854, "before_s": 279, "after_s": 53}
print(f"{r['checks']} checks; found Rs {r['undercharged']:,} undercharged and a Rs {r['upi_gap']:,} UPI gap; "
      f"clean job {r['before_s']}s -> {r['after_s']}s ({r['before_s'] / r['after_s']:.1f}x)")

4 days × ₹2,500 · ₹43,710 · ₹6,854

Rs 60,564 in the first month (illustrative)

time_saved = 4 * 2500
print(f"Rs {time_saved + 43710 + 6854:,} in the first month (illustrative)")

components

1. kill a map task JVM    -> attempt re-runs, counters exact
2. stop a DataNode        -> reads use replicas; writes continue (policy on small clusters)
3. stop a NodeManager     -> containers lost after expiry, tasks rescheduled
4. fill a disk to 97%     -> write failures; alert before it happens (Lesson 13.4 incident)
5. re-run a finished day  -> idempotent: totals unchanged

drills = [("kill a map task JVM", "attempt re-runs, counters exact"),
          ("stop a DataNode", "reads use replicas; writes continue (policy on small clusters)"),
          ("stop a NodeManager", "containers lost after expiry, tasks rescheduled"),
          ("fill a disk to 97%", "write failures; alert before it happens (Lesson 13.4 incident)"),
          ("re-run a finished day", "idempotent: totals unchanged")]
for i, (what, observe) in enumerate(drills, 1):
    print(f"{i}. {what:<22} -> {observe}")

Real-world scenario questions (asked in company interviews)

Project deep-dive questions, the kind asked for 20–30 minutes in data-engineering interviews.