Hadoop Lakehouse Migration: HDFS to Object Store, Hive to Iceberg, Job Rewrites

A Hadoop migration is never one project — it is four migrations stacked on top of each other, and the teams that treat it as a single "lift the cluster to the cloud" effort are the ones that stall for eighteen months and quietly move back. The storage layer (HDFS) has to become an object store; the table metadata (the Hive metastore and its directory-listing tables) has to become an open table format like Iceberg; the compute (MapReduce, Tez, Hive-on-YARN, Spark-on-YARN) has to become jobs that run against elastic compute; and every one of those moves has to happen while the old estate keeps serving production, with a reconciliation story and a rollback plan for each table. Get the sequencing wrong — rewrite the jobs before the data lands, or cut over before you reconcile — and you ship a silent correctness bug into a warehouse that a hundred downstream dashboards trust.

This guide is the senior-data-engineering walkthrough for the lakehouse migration you will actually be asked to lead or to whiteboard: the HDFS to S3 bulk-and-incremental copy with the S3A committers that fix the object-store rename problem, the Hive to Iceberg cutover that adopts your existing Parquet without a full rewrite, the job rewrite from HiveQL and MapReduce onto Spark migration targets with the semantic traps that break parity, and the dual-run cutover that ends in a real cluster decommission once the numbers match. It covers each layer from the angle interviewers probe — how you copy petabytes without saturating the link, why object storage has no atomic rename, how Iceberg turns a directory of files into a snapshot log, and when you are actually allowed to power the DataNodes off — and each section pairs a teaching block with a Solution-Tail interview answer: code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse the transformation reps on the data-processing practice library →, and stress-test the tuning fundamentals on the optimization practice library →.

---

On this page

• Why the Hadoop → Lakehouse migration reshapes the whole platform
• HDFS to object store — the storage-layer migration
• Hive to Iceberg — the table-format migration
• Job rewrites — MapReduce / HiveQL / Pig → Spark
• Cutover, reconciliation & cluster decommission
• Cheat sheet — Hadoop → lakehouse migration recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why the Hadoop → Lakehouse migration reshapes the whole platform

Four layers that move independently — and a sequencing mistake in any one of them ships a correctness bug

The one-sentence invariant: a Hadoop migration is the coordinated retirement of four independent layers — HDFS storage, Hive-metastore table metadata, MapReduce/YARN compute, and Oozie-style orchestration — each of which migrates on its own schedule onto a decoupled object-store-plus-open-table-format-plus-elastic-compute stack, and the entire risk of the project lives in the seams between those layers rather than in any single layer's mechanics. Anyone can run one distcp. The hard part is that the data copy, the table-format switch, the job rewrite, and the cutover each has a different failure mode, a different reconciliation story, and a different rollback, and they have to be interleaved so that no downstream consumer ever reads a half-migrated table. The team that copies the data, rewrites the jobs, and flips everything on one weekend has no way to tell whether a row-count drift came from the copy, the format, or the rewrite — so they cannot debug it, and they roll back the whole thing.

The four axes interviewers actually probe.

• Data-copy strategy. How do you move petabytes off HDFS without saturating the network link or taking an outage? The senior answer names DistCp for the bulk copy, an incremental pass (-update / -diff against snapshots) to catch the delta accumulated during the long bulk run, bandwidth throttling so you do not starve production, and the fact that the object store has no atomic rename, which changes how jobs commit output. Weak candidates say "we'd copy it to S3" and stop.
• Metadata / table-format continuity. The Hive metastore is not just a schema catalog — it is the thing every query resolves partitions against. Moving to Iceberg means the catalog changes (Glue, a REST catalog, Nessie, or a metastore-backed Iceberg catalog) and the on-disk table representation changes from "a directory whose subfolders are partitions" to "a metadata tree of manifests and snapshots." The senior answer knows you can adopt existing Parquet files without rewriting them via Iceberg's snapshot / add_files procedures.
• Job-rewrite blast radius. How many jobs, in what languages, with what semantic differences? HiveQL is mostly Spark-SQL-compatible but the edge cases (implicit casts, NULL versus empty string, decimal precision, LATERAL VIEW, reserved words) are exactly where parity breaks. MapReduce and Pig are full rewrites. Oozie becomes Airflow. The senior answer scopes the rewrite by counting jobs and classifying them, then proves each port with a golden-output diff rather than eyeballing.
• Cutover, rollback & decommission. You never big-bang. You dual-run — the legacy Hadoop pipeline and the new lakehouse pipeline both produce the table, you reconcile them, and only when the numbers match for N consecutive runs do you cut the downstream consumers over. Every wave has a rollback (repoint the consumer at the Hadoop output). Decommission is the last step — drain YARN, retire the DataNodes, delete HDFS — and it is what actually captures the cost saving that justified the project.

The 2026 reality — decouple storage from compute, and Iceberg is the default open table format.

• Storage and compute are separate now. The whole point of the lakehouse is that HDFS's co-location of storage and compute on the same DataNodes — the thing that made Hadoop fast in 2012 — is now a liability. Object storage scales and is priced independently; compute (Spark on Kubernetes/EMR/Dataproc/serverless) autoscales to zero. You stop paying for a cluster that is idle at 3 a.m.
• Iceberg (or Delta/Hudi) replaces the Hive table format. The Hive "table = directory, partition = subdirectory" model has no atomic commit, no schema evolution by column identity, and a metastore that melts on tables with millions of partitions. Iceberg replaces directory listing with a manifest tree, adds snapshots (time travel, atomic commit, rollback), hidden partitioning, and safe schema evolution — while keeping your data as ordinary Parquet.
• Spark / Trino replace MapReduce and Tez. MapReduce is effectively deprecated. HiveQL still runs, but on Spark or Trino as the engine, not on MapReduce/Tez. Most migrations standardise on Spark for batch ETL and Trino for interactive SQL, both reading the same Iceberg tables.
• Orchestration moves to Airflow / Dagster. Oozie's XML workflows become Python DAGs. This is usually the least risky layer to move because it is a rewrite of scheduling, not of data semantics.

What interviewers listen for.

• Do you decompose the migration into four independent layers rather than "move Hadoop to the cloud"? — senior signal.
• Do you say "never big-bang; dual-run and reconcile" before you are asked about validation? — required answer.
• Do you know Iceberg can adopt existing Parquet without a rewrite (snapshot / add_files)? — senior signal.
• Do you name the object store's lack of atomic rename as the reason job commit changes? — senior signal.
• Do you make decommission the last step and tie it to the cost model that justified the project? — required answer.

Worked example — the four-layer migration map

Detailed explanation. The single most useful artifact for a Hadoop-migration interview is a map that separates the four layers, states the from and to of each, and names the tool and the reconciliation for each. Every senior migration discussion converges on this decomposition within the first ten minutes; having it in your head is what turns a rambling "we'd move it to the cloud" into a fluent plan.

• Storage layer. HDFS (block-replicated, co-located) → object store (S3 / ADLS / GCS, flat namespace, decoupled).
• Table-format layer. Hive metastore + directory tables → Iceberg tables in a catalog (Glue / REST / Nessie).
• Compute layer. MapReduce / Tez / Hive-on-YARN / Spark-on-YARN → Spark (batch) + Trino (interactive) on elastic compute.
• Orchestration layer. Oozie XML → Airflow / Dagster DAGs.

Question. Lay out the four-layer map for a 3 PB Hadoop estate with 1,400 Hive tables and ~600 scheduled jobs, and name the tool and the reconciliation check for each layer.

Input.

Layer	From	To	Tool
Storage	HDFS (3× replication)	S3 (object store)	DistCp + S3A committer
Table format	Hive metastore tables	Iceberg tables	snapshot / migrate / add_files
Compute	MapReduce / HiveQL / Spark-on-YARN	Spark + Trino	job rewrite + golden diff
Orchestration	Oozie	Airflow	DAG rewrite

Code.

Hadoop → Lakehouse — the four-layer map (fill this in per estate)
=================================================================

LAYER          FROM                         TO                         VALIDATION
-----          ----                         --                         ----------
storage        HDFS /warehouse/*            s3://lake/warehouse/*      distcp -diff, byte counts
table format   Hive metastore (dir tables)  Iceberg @ Glue catalog     row counts + snapshot id
compute        MapReduce / HiveQL / Pig     Spark SQL / Spark / Trino  golden-output diff
orchestration  Oozie coordinators           Airflow DAGs               same-schedule dry run

SEQUENCING RULE
  1. copy storage first (data must exist before jobs run on it)
  2. adopt table format on the copied data (Iceberg over the Parquet)
  3. rewrite + dual-run jobs against the new tables
  4. reconcile, cut over table-by-table, THEN decommission

DO NOT
  - rewrite jobs before the data has landed
  - cut over a table before N clean reconcile runs
  - decommission anything until every dependent consumer is moved

Step-by-step explanation.

1. The four layers are listed in dependency order: storage must exist before a table format can wrap it, a table must exist before a job can read it, and a job must be validated before a consumer can be cut over. This ordering is the backbone of the whole plan.
2. Each layer gets its own tool and its own validation. Storage is validated by byte/row counts after distcp; the table format by comparing row counts and recording the Iceberg snapshot id; compute by a golden-output diff; orchestration by a same-schedule dry run. Mixing validations across layers is what makes a drift undiagnosable.
3. Storage moves first because everything downstream reads it. You copy the Parquet as-is — no transformation — so that the copy is a pure byte-for-byte movement you can checksum.
4. The table format adopts the already-copied data. Because Iceberg can point at existing Parquet, this step rewrites metadata, not data — cheap and fast relative to the copy.
5. Jobs are rewritten and dual-run only after their input tables exist in the new format. Decommission is deliberately last and gated on every consumer being moved — it is the irreversible step.

Output.

Sequencing decision	Why it is in this position
Copy storage first	jobs cannot run on data that has not landed
Adopt Iceberg on copied data	metadata-only; no second data rewrite
Rewrite + dual-run jobs	needs the new tables as input
Reconcile before cutover	proves parity per table
Decommission last	irreversible; capture the cost saving

Rule of thumb. Draw the four-layer map before you write a single distcp. Storage → table format → compute → orchestration, each with its own tool and its own reconciliation, migrated in dependency order, with decommission gated on the last consumer moving. The plan falls out of the map.

Worked example — the wave plan (never big-bang)

Detailed explanation. A 1,400-table estate does not migrate in one cutover; it migrates in waves, where each wave is a bundle of tables plus the jobs that produce and consume them, chosen so the wave is internally consistent (you never split a producer from its consumer across waves). The wave plan is what makes the migration incremental, reversible, and measurable. Walk through slicing the estate into waves.

• Wave sizing. 50–150 tables per wave, grouped by data domain (finance, clickstream, catalog) so a wave's producers and consumers stay inside the wave.
• Wave ordering. Lowest-risk / lowest-dependency domains first (a reporting mart with few upstreams), highest-blast-radius last (the core fact tables everything joins to).
• Per-wave lifecycle. copy → adopt → dual-run → reconcile N runs → cut over consumers → keep legacy warm for a rollback window → retire the wave's HDFS paths.

Question. Slice the estate into waves and define the entry and exit criteria for one wave.

Input.

Wave	Domain	Tables	Risk	Order
1	Reporting marts	120	low (leaf consumers)	first
2	Clickstream	300	medium	second
3	Catalog / dims	200	medium	third
4	Core finance facts	180	high (everything joins)	last

Code.

Wave exit criteria (a wave is "done" only when ALL are true)
============================================================
[ ] every table in the wave exists as an Iceberg table on the object store
[ ] every producing job is rewritten and green on the new stack
[ ] dual-run reconciliation is CLEAN for >= 7 consecutive daily runs
      row_count_delta == 0
      sum(amount) delta within 0.00 (exact) for money columns
      partition-level counts match
[ ] every downstream consumer repointed to the Iceberg table
[ ] rollback tested: repoint one consumer back to legacy, confirm it works
[ ] legacy output kept warm for the rollback window (e.g. 14 days)

Only after the rollback window closes for ALL waves:
[ ] drain YARN queues, retire NodeManagers
[ ] delete HDFS warehouse paths (after final checksum archive)
[ ] power down DataNodes  -> cost saving realised here

Step-by-step explanation.

1. Waves are grouped by data domain so a wave is self-contained: the jobs that produce a wave's tables and the jobs that consume them live in the same wave. Splitting a producer from its consumer would force a cross-stack join mid-migration.
2. Ordering runs low-risk-first. Reporting marts are leaves — few things depend on them — so a mistake there is contained. Core finance facts are last because every join touches them; you want maximum practice before you touch the highest-blast-radius tables.
3. The exit criteria are conjunctive — every box must be checked. The reconciliation gate ("clean for ≥ 7 consecutive runs") is the one that prevents a subtle drift from slipping through on a lucky single run.
4. The rollback window keeps the legacy output warm after cutover, so if a consumer reports a problem three days later you repoint it back to the Hadoop output in minutes. This is the safety net that makes each wave reversible.
5. Decommission is outside every wave — it happens once, globally, only after the last wave's rollback window closes. That is the single irreversible act and the moment the cost model pays off.

Output.

Wave	Entry gate	Exit gate
1 Reporting	inputs copied	7 clean reconciles + consumers moved
2 Clickstream	wave 1 stable	7 clean reconciles + consumers moved
3 Catalog	wave 2 stable	7 clean reconciles + consumers moved
4 Core facts	waves 1–3 stable	7 clean reconciles + consumers moved
(global) Decommission	all waves past rollback window	HDFS deleted, DataNodes off

Rule of thumb. Migrate in domain-aligned waves, low-risk first, with conjunctive exit criteria and a warm-legacy rollback window per wave. Decommission is a single global step gated on the last wave — never fold it into a wave.

Worked example — what interviewers actually probe

Detailed explanation. The senior Hadoop-migration interview has a predictable arc: an ambiguous opener ("we have a big on-prem Hadoop cluster, how would you get us to the cloud?"), then progressive narrowing to test whether you know the four layers and the cutover discipline. Candidates who decompose into layers and reach for dual-run-and-reconcile score highest; candidates who say "we'd rehost the cluster on cloud VMs" score lowest. Walk through the grading rubric.

• Ambiguous opener. "How would you migrate our Hadoop platform to the cloud?" — invites the four-layer decomposition.
• Follow-up 1. "How do you move 3 PB off HDFS?" — probes the data-copy strategy.
• Follow-up 2. "Do you have to rewrite all the Parquet to use Iceberg?" — probes table-format knowledge.
• Follow-up 3. "How do you know the migrated table is correct?" — probes reconciliation.
• Follow-up 4. "When can you turn the cluster off?" — probes cutover + decommission discipline.

Question. Draft a five-minute senior migration answer that covers all four layers and the cutover discipline without waiting for the follow-ups.

Input.

Interview signal	Weak answer	Senior answer
Framing	"rehost the cluster on cloud VMs"	"four independent layers: storage, table format, compute, cutover"
Data copy	"copy it to S3"	"DistCp bulk + incremental -diff, throttled, S3A committer"
Table format	"reload everything into Iceberg"	"adopt existing Parquet via snapshot/add_files; no data rewrite"
Validation	"we'd test it"	"dual-run + reconcile row counts, sums, partitions for N runs"
Decommission	"then we shut down Hadoop"	"decommission last, gated on last consumer + rollback window"

Code.

Senior Hadoop-migration answer template (5 minutes)
===================================================

Minute 1 — decompose into four layers
  "This isn't one migration, it's four: HDFS storage, Hive-metastore
   table format, MapReduce/YARN compute, and Oozie orchestration. Each
   moves on its own schedule; the risk is in the seams."

Minute 2 — storage
  "Bulk-copy HDFS to the object store with DistCp, then an incremental
   -diff pass to catch the delta from the long bulk run. Throttle
   bandwidth so we don't starve prod. Object stores have no atomic
   rename, so jobs must commit via the S3A magic committer or Iceberg."

Minute 3 — table format
  "Adopt Iceberg over the copied Parquet WITHOUT rewriting data —
   snapshot for a read-only shadow, add_files/migrate for in-place.
   That buys snapshots, ACID commit, schema evolution, hidden
   partitioning, and time-travel rollback."

Minute 4 — compute + validation
  "Rewrite jobs onto Spark/Trino. HiveQL is mostly Spark-SQL-compatible
   but I pin the semantic traps — implicit casts, NULL vs empty string,
   decimal precision — and prove each port with a golden-output diff.
   Then dual-run: both stacks produce the table, reconcile row counts,
   money sums, and partition counts for >= 7 clean runs."

Minute 5 — cutover + decommission
  "Cut over table-by-table in domain waves, keep legacy warm for a
   rollback window, then — and only then — drain YARN, retire the
   DataNodes, delete HDFS. Decommission is the last step and it's where
   the cost saving is actually realised."

Step-by-step explanation.

1. Minute 1 is the framing that scores. Naming four independent layers immediately signals you understand that the risk is in the seams, not in any one tool. "Rehost on cloud VMs" is the answer that loses the room because it moves the liability instead of removing it.
2. Minute 2 pre-empts the data-copy follow-up and, crucially, names the no-atomic-rename property of object stores — the single fact that separates people who have run a cloud migration from people who have only read about one.
3. Minute 3 shows you know Iceberg adopts existing Parquet without a rewrite. Candidates who say "reload everything into Iceberg" are quoting a data copy you do not need and cannot afford at petabyte scale.
4. Minute 4 splits the compute answer into rewrite and validation, and names the semantic traps explicitly. "We'd test it" loses; "golden-output diff plus dual-run reconciliation for ≥ 7 clean runs" wins.
5. Minute 5 makes decommission the last step and ties it to the cost model. This is the discipline the interviewer is listening for: the irreversible act is gated, and it is the act that justifies the entire project financially.

Output.

Grading criterion	Weak score	Senior score
Decomposes into four layers	rare	mandatory
Names DistCp + incremental + committer	rare	senior signal
Adopts Parquet without rewrite	rare	senior signal
Dual-run + reconcile before cutover	occasional	required
Decommission last, cost-tied	rare	senior signal

Rule of thumb. The senior Hadoop-migration answer is a five-minute monologue: four independent layers, DistCp-plus-committer for storage, adopt-Iceberg-without-rewrite for the table format, rewrite-and-reconcile for compute, and a gated decommission last. Rehearse it once; deploy it every interview.

Senior interview question on migration sequencing

A senior interviewer often opens with: "You inherit a 3 PB, 1,400-table, 600-job on-prem Hadoop estate and a mandate to be off it in a year. Leadership wants a date for turning the cluster off. Walk me through how you sequence the four layers, how you slice the estate so it is never half-migrated from a consumer's point of view, and how you decide — with evidence — when it is safe to decommission."

Solution Using a dependency-ordered, wave-based plan with a gated decommission

# 1. SEQUENCE the four layers in dependency order (per wave)
#    storage -> table format -> compute -> orchestration -> cutover
copy_storage        (DistCp bulk + snapshot -diff, S3A committer)
adopt_table_format  (Iceberg snapshot/migrate over the copied Parquet)
rewrite_jobs        (Spark/Trino + golden-output diff, dual-run)
port_orchestration  (Oozie -> Airflow, same-schedule dry run)
reconcile_and_cut   (tiered reconcile, N clean runs, view repoint)

# 2. SLICE into domain-aligned waves so no producer is split from its
#    consumer, low-risk domains first, core facts last.
waves = [
    {"name": "reporting", "tables": 120, "risk": "low",    "order": 1},
    {"name": "clickstream","tables": 300, "risk": "medium", "order": 2},
    {"name": "catalog",    "tables": 200, "risk": "medium", "order": 3},
    {"name": "core_facts", "tables": 180, "risk": "high",   "order": 4},
]

def wave_done(w) -> bool:                       # conjunctive exit criteria
    return (w["adopted"] and w["jobs_green"]
            and w["reconcile_clean_runs"] >= 7
            and w["consumers_repointed"]
            and w["rollback_tested"])

# 3. DECOMMISSION only when every wave is past its rollback window
def safe_to_decommission(waves, rollback_days=14) -> bool:
    return all(wave_done(w) and w["days_since_cutover"] > rollback_days
               for w in waves)

Step-by-step trace.

Stage	Gate	Evidence produced
Copy storage	snapshot -diff clean	per-prefix counts + sizes
Adopt Iceberg	exact count + integer sum	snapshot id, parity report
Rewrite jobs	golden-output diff PASS	byte-parity per job
Reconcile	7 consecutive clean runs	reconcile history
Cutover	consumers repointed + rollback tested	view target = Iceberg
Decommission	every wave past rollback window	no hdfs:// references

Sequencing storage-first means jobs always have data to run on; slicing by domain keeps every producer and consumer inside one wave, so a consumer never straddles the two stacks; and the conjunctive per-wave exit criteria plus the "past the rollback window" gate turn "are we safe to turn it off?" from a judgement call into a checklist that produces evidence. The date leadership wants is the date the last wave clears its rollback window — not a wishful cutover weekend.

Output:

Question from leadership	Evidence-based answer
Is it half-migrated?	no — waves are domain-complete
How do we know it's correct?	7 clean reconciles + golden diffs per wave
Can we roll back?	yes, within each wave's window
When can we turn it off?	when the last wave clears its window
Where's the saving?	realised at DataNode power-off

Why this works — concept by concept:

• Dependency-ordered sequencing — storage → table format → compute → orchestration → cutover is the only order where each step's inputs already exist. Rewriting jobs before data lands, or cutting over before reconciling, are the two sequencing mistakes that make a drift undiagnosable.
• Domain-aligned waves — grouping tables by data domain keeps every producer with its consumers, so a wave is internally consistent and no downstream reader ever sees a half-migrated join. Low-risk-first buys practice before the high-blast-radius core facts.
• Conjunctive exit criteria — a wave is done only when adoption, job parity, N clean reconciles, consumer repointing, and a tested rollback are all true. Any single unchecked box holds the wave, which is what stops an optimistic cutover.
• Rollback-window gate on decommission — the irreversible step is gated on every wave being past its warm-legacy window, so "safe to turn off" is a checklist producing evidence, not a gut call.
• Cost — the plan is O(waves) sequential effort with a temporary dual-run overlap where you pay for both stacks; the saving is O(0) idle-compute spend but only realised at power-off. Time-boxing the windows bounds the overlap cost, which is why the decommission date is the real project deadline.

ETL
Topic — etl
ETL problems on large-scale data migration

Practice →

Design Topic — design Design problems on platform migration and cutover

Practice →

---

2. HDFS to object store — the storage-layer migration

HDFS to S3 is a bulk copy plus an incremental catch-up — and the object store's missing atomic rename is what actually breaks your jobs

The mental model in one line: moving object storage under a Hadoop estate is a two-phase copy — a long DistCp bulk transfer of the whole warehouse followed by an incremental -diff pass that catches everything written during the bulk run — layered on top of one hard truth: an object store has a flat namespace with no atomic directory rename, so the Hadoop commit protocol that renames a _temporary directory into place at the end of a job is unsafe, and you must switch to an S3A committer (or a table format like Iceberg) that never relies on rename. Every senior data engineer who has done this migration has been burned once by a job that "succeeded" but left partial output, because they ran the classic FileOutputCommitter against S3.

The four axes for the storage migration.

• Copy strategy. Bulk first (distcp the entire warehouse), then incremental (-update to copy only changed files, or -diff between two HDFS snapshots to copy exactly the delta). The bulk run can take days at petabyte scale; the incremental pass closes the gap that accumulated while it ran.
• Consistency & rename. Modern S3 is strongly consistent for read-after-write (since late 2020), so the old "eventual consistency" hazard is gone — but the rename hazard is not. Object stores implement "rename" as copy-then-delete, which is neither atomic nor cheap. The Hadoop v1/v2 FileOutputCommitter renames task output into place to commit; on an object store that rename can partially fail, leaving a job that reports success with missing files.
• Layout & locality. HDFS gives you data locality (compute runs on the node holding the block) and directory semantics. Object stores give you neither: there is no locality (compute reads over the network), and "directories" are just key prefixes. Small-file problems get worse because every object is a separate GET; you want fewer, larger files (compaction) after the move.
• Cost & elasticity. This is the prize. Storage is now billed per GB-month independent of any cluster, and compute autoscales to zero. You stop paying for 200 always-on DataNodes to hold cold data. The migration's business case is almost entirely this line item.

DistCp — the bulk workhorse.

• What it is. A MapReduce job that copies files in parallel from a source filesystem to a destination filesystem. Each mapper copies a slice of the file list; throughput scales with mapper count up to the network ceiling.
• Bulk invocation. hadoop distcp -m <mappers> -bandwidth <MB/s> hdfs://nn/warehouse s3a://lake/warehouse — throttle -bandwidth per mapper so the aggregate does not starve production traffic.
• Incremental. -update copies only files whose size/checksum differ; -diff snap1 snap2 uses HDFS snapshots to copy exactly the files that changed between two points — the correct tool for the catch-up pass after a multi-day bulk run.
• Verification. distcp can compare checksums, but HDFS (CRC32C over blocks) and S3 (ETag / multipart) use different checksum schemes, so cross-filesystem checksum comparison is unreliable — verify with counts and sizes, and for critical tables a content-level reconciliation (next section).

The rename problem and S3A committers.

• Why rename matters. Spark/MapReduce write task output to a staging path, then commit by renaming staging → final. On HDFS rename is an atomic metadata operation. On S3 it is a copy of every object plus a delete — slow, non-atomic, and able to leave the final path half-populated if it fails midway.
• The magic committer. fs.s3a.committer.name=magic uses S3 multipart uploads: task output is uploaded but not completed until job commit, at which point the multipart uploads are finalised — an atomic-per-file completion with no rename.
• The directory committer. Stages task output on the local disk / HDFS and uploads on commit; simpler, good when you have a small HDFS still available during migration.
• Or sidestep it entirely. Iceberg (next section) commits by writing a new metadata file and swapping a single catalog pointer — it never renames data files, so the rename problem disappears the moment your tables are Iceberg.

Common interview probes on the storage migration.

• "How do you copy 3 PB off HDFS?" — DistCp bulk + incremental -diff, throttled, verified by counts.
• "What's different about writing to S3 versus HDFS?" — no atomic rename; use an S3A committer or Iceberg.
• "Is S3 eventually consistent?" — no longer; strong read-after-write since 2020, but rename is still copy+delete.
• "What gets worse after the move?" — small files (every object is a GET) and loss of data locality; compact after copy.

Worked example — DistCp bulk copy plus an incremental catch-up

Detailed explanation. The canonical storage move: a bulk distcp of the whole warehouse that runs for days, then an incremental -diff pass driven by HDFS snapshots that copies exactly the files written while the bulk job ran. This closes the gap without re-copying petabytes. Walk through both passes.

• Bulk. distcp the full /warehouse prefix with a mapper count tuned to the link and a per-mapper bandwidth cap.
• Snapshots. Take an HDFS snapshot before the bulk run; take a second after; -diff copies only the delta.
• Verify. Compare file counts and total bytes per top-level table prefix.

Question. Write the bulk copy, the snapshot-driven incremental catch-up, and the count/size verification.

Input.

Parameter	Value
Source	hdfs://nn/warehouse
Destination	s3a://lake/warehouse
Mappers	200
Per-mapper bandwidth	20 MB/s (≈ 4 GB/s aggregate)
Delta strategy	HDFS snapshot -diff

Code.

# 0. Enable snapshots on the source dir and take a "before" snapshot
hdfs dfsadmin -allowSnapshot /warehouse
hdfs dfs -createSnapshot /warehouse snap_before

# 1. BULK copy — the multi-day run of the whole warehouse
hadoop distcp \
  -Dmapreduce.job.name=distcp-warehouse-bulk \
  -m 200 \
  -bandwidth 20 \
  -update \
  -strategy dynamic \
  hdfs://nn/warehouse \
  s3a://lake/warehouse

# 2. After bulk completes, take an "after" snapshot ...
hdfs dfs -createSnapshot /warehouse snap_after

# 3. INCREMENTAL catch-up — copy ONLY the delta between the two snapshots
#    (files created/changed while the bulk job was running)
hadoop distcp \
  -Dmapreduce.job.name=distcp-warehouse-delta \
  -m 100 \
  -bandwidth 20 \
  -update \
  -diff snap_before snap_after \
  hdfs://nn/warehouse \
  s3a://lake/warehouse

# 4. Verify per table prefix — counts and bytes (checksums differ across FS,
#    so compare counts + sizes, then content-reconcile critical tables)
for tbl in orders customers shipments; do
  src=$(hdfs dfs -count -q /warehouse/$tbl | awk '{print $2" files "$3" bytes"}')
  dst=$(hadoop fs -count -q s3a://lake/warehouse/$tbl | awk '{print $2" files "$3" bytes"}')
  echo "$tbl  HDFS: $src   S3: $dst"
done

Step-by-step explanation.

1. Step 0 enables HDFS snapshots and freezes a snap_before marker. Snapshots are cheap copy-on-write metadata pointers — they do not duplicate data — and they are what makes the later -diff exact rather than a guess based on modification times.
2. Step 1 is the bulk distcp. -m 200 runs 200 parallel mappers; -bandwidth 20 caps each mapper at 20 MB/s so the aggregate (~4 GB/s) leaves headroom for production. -strategy dynamic hands work to mappers as they finish, so a few huge files do not leave most mappers idle. -update makes the copy idempotent — a retried run skips files already present with matching size.
3. Step 2 takes snap_after once the bulk run finishes. The delta between snap_before and snap_after is precisely the set of files that changed during the (possibly multi-day) bulk copy — new partitions, late-arriving files, compactions.
4. Step 3's -diff snap_before snap_after tells distcp to copy exactly that delta and nothing else. Without this, the catch-up would either re-scan the entire warehouse (slow) or rely on -update timestamp heuristics (error-prone). The snapshot diff is the correct, exact mechanism.
5. Step 4 verifies with counts and bytes because HDFS and S3 use different checksum algorithms — a cross-filesystem checksum comparison produces false mismatches. Counts and sizes catch gross errors; a content-level reconciliation (Section 5) catches subtle ones for the tables that matter.

Output.

Pass	Files copied	Bytes	Duration
Bulk (snap_before)	42,000,000	3.0 PB	~62 h
Incremental (-diff)	310,000	21 TB	~1.4 h
Verify	counts + sizes match per prefix	—	minutes

Rule of thumb. Bulk-copy with distcp -update -strategy dynamic, throttle per-mapper bandwidth to protect production, and drive the catch-up with an HDFS snapshot -diff rather than timestamp heuristics. Verify with counts and sizes, never cross-filesystem checksums.

Worked example — configuring the S3A committer to survive the missing rename

Detailed explanation. The moment a Spark job writes output to S3 with the default committer, you are exposed: the commit renames a _temporary directory into place, and on S3 that "rename" is a non-atomic copy+delete. The fix is the S3A magic committer, which uses multipart uploads and completes them at job-commit time — no rename. Walk through the config and what changes.

• The hazard. Default FileOutputCommitter (v1 or v2) commits by renaming staging output into the final path.
• The fix. fs.s3a.committer.name=magic + the committer factory, so Spark commits via multipart-upload completion.
• The proof. A failed task leaves no partial files in the final path, because uploads are only completed on successful job commit.

Question. Configure Spark to use the S3A magic committer and explain what each setting prevents.

Input.

Setting	Value	Purpose
committer name	magic	avoid rename-based commit
committer factory	S3A factory	route commit through S3A
conflict mode	replace	overwrite partition on rerun
default FileOutputCommitter	disabled	do not fall back to rename

Code.

# Spark session configured for safe S3 writes (magic committer, no rename)
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("lakehouse-write")
    # Route all commits through the S3A committer factory ...
    .config(
        "spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a",
        "org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory",
    )
    # ... and pick the MAGIC committer (multipart upload; no rename)
    .config("spark.hadoop.fs.s3a.committer.name", "magic")
    .config("spark.hadoop.fs.s3a.committer.magic.enabled", "true")
    # On rerun, replace the target partition rather than appending duplicates
    .config("spark.hadoop.fs.s3a.committer.staging.conflict-mode", "replace")
    # Parquet must use the committer's output-committer, not the rename one
    .config(
        "spark.sql.parquet.output.committer.class",
        "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter",
    )
    .config(
        "spark.sql.sources.commitProtocolClass",
        "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol",
    )
    .getOrCreate()
)

# This write now commits via multipart completion — a failed task leaves
# NO partial files in the final path.
(
    spark.read.parquet("s3a://lake/warehouse/orders")
    .where("order_date = '2026-08-17'")
    .write.mode("overwrite")
    .parquet("s3a://lake/warehouse/orders_daily/2026-08-17")
)

Step-by-step explanation.

1. The committer factory config routes every S3A output commit through Hadoop's S3A committer machinery instead of the default FileOutputCommitter. Without this line, the other settings are ignored and Spark silently falls back to rename-based commit.
2. fs.s3a.committer.name=magic selects the magic committer, which writes task output as incomplete S3 multipart uploads directly to the final key, and only completes those uploads during job commit. Completion is atomic per object, so there is no window where a half-written directory is visible.
3. conflict-mode=replace makes a rerun overwrite the target partition rather than layering a second copy on top — essential during dual-run when you will re-execute the same day repeatedly.
4. The two Parquet/commit-protocol classes bind Spark's SQL writer to the path-output committer so that even DataFrame .write.parquet(...) (not just RDD saves) goes through the safe path. Missing these is the classic "I set the committer but Spark still renamed" bug.
5. The payoff: a task that dies mid-write never completes its multipart uploads, so the final path contains only fully-committed files. You have replaced HDFS's atomic-rename guarantee with multipart-completion atomicity — the correct object-store analogue.

Output.

Scenario	Default FileOutputCommitter on S3	S3A magic committer
Task fails mid-write	partial files may remain	no partial files
Commit mechanism	copy + delete (rename)	multipart completion
Commit cost	O(bytes) copy	O(1) metadata complete
Job "succeeds" but output missing	possible	prevented

Rule of thumb. Never write to an object store with the default FileOutputCommitter. Bind the S3A magic committer (factory + name + Parquet/commit-protocol classes) so commit is multipart completion, not rename — or move the table to Iceberg and let the catalog pointer be your atomic commit.

Worked example — decoupling storage from compute and the cost model

Detailed explanation. The number that justifies the whole migration is the storage/compute decoupling. On HDFS you pay for DataNodes 24×7 to hold data whether or not anything is querying it. On the lakehouse you pay for object storage per GB-month and for compute only while a job runs. Walk through the cost model that turns "we should modernise" into a signed-off budget.

• Before. N always-on DataNodes = storage + compute fused; idle cluster still bills.
• After. Object storage priced per GB-month; compute (Spark on K8s/EMR/serverless) autoscales to zero between jobs.
• The saving. Cold data no longer requires running machines; peak compute is provisioned only at peak.

Question. Model the monthly cost before and after for a 3 PB / 200-DataNode estate whose cluster is busy ~30% of the day.

Input.

Component	Before (HDFS)	After (lakehouse)
Storage	200 DataNodes (fused)	3 PB object storage @ per-GB-month
Compute	same 200 nodes, 24×7	autoscaled Spark, ~30% duty
Idle cost	full cluster runs at 3 a.m.	compute scales to ~0
Replication overhead	3× on-disk	erasure-coded in object store

Code.

# Illustrative cost model — before vs after (numbers are placeholders;
# plug in your own rates)
PB = 3
GB = PB * 1024 * 1024                     # 3 PB in GB

# BEFORE: 200 DataNodes, always on (storage + compute fused)
NODES            = 200
NODE_HOURLY      = 1.20                    # $/node-hour (HW amortised + power + DC)
HOURS_MONTH      = 730
before_monthly   = NODES * NODE_HOURLY * HOURS_MONTH

# AFTER: object storage per GB-month + compute only while jobs run
OBJ_PER_GB_MONTH = 0.021                   # $/GB-month (single copy; EC not 3x)
storage_after    = GB * OBJ_PER_GB_MONTH

COMPUTE_NODES    = 200                     # peak, but only ~30% of the time
COMPUTE_HOURLY   = 1.50                    # cloud rate incl. spot mix
DUTY             = 0.30                    # busy 30% of the day (autoscale off-peak)
compute_after    = COMPUTE_NODES * COMPUTE_HOURLY * HOURS_MONTH * DUTY

after_monthly    = storage_after + compute_after

print(f"before:  ${before_monthly:,.0f}/mo  (fused, 24x7)")
print(f"after:   ${after_monthly:,.0f}/mo  (storage ${storage_after:,.0f} + compute ${compute_after:,.0f})")
print(f"saving:  ${before_monthly - after_monthly:,.0f}/mo  "
      f"({100*(before_monthly-after_monthly)/before_monthly:.0f}% lower)")

Step-by-step explanation.

1. The "before" line fuses storage and compute: 200 DataNodes bill for all 730 hours in the month regardless of utilisation, because HDFS keeps the data resident on running machines. The 3 a.m. idle cluster still costs full price.
2. The "after" line splits the two. Storage becomes a flat per-GB-month charge on a single erasure-coded copy — you shed the 3× replication overhead HDFS imposed, so effective stored bytes drop even before the price difference.
3. Compute is billed only for the ~30% duty cycle: autoscaling drops executors to near zero off-peak, and a spot/preemptible mix trims the peak rate. The fused model could never do this because the storage could not be separated from the compute.
4. The subtraction is the business case. The saving is dominated by not paying for idle compute and by dropping 3× replication — both direct consequences of decoupling, neither available while storage lives on DataNodes.
5. This model is also the decommission trigger: the saving is only realised when the DataNodes are actually powered off (Section 5). Until then you are paying for both stacks, which is why the rollback window is time-boxed.

Output.

Line	Before	After
Storage	included in node cost	flat per-GB-month, single copy
Compute	24×7	~30% duty, autoscaled
Idle spend	full cluster	near zero
Realised when	—	DataNodes powered off

Rule of thumb. The migration's business case is the storage/compute split: you stop paying for idle compute and for 3× replication. Build the before/after model early, and remember the saving is only realised at decommission — so keep the dual-run rollback window time-boxed.

Senior interview question on the storage-layer migration

A senior interviewer often opens with: "You need to move a 3 PB HDFS warehouse to S3 while the cluster keeps serving production, with zero data loss and no multi-day read outage. Walk me through the bulk-plus-incremental copy, how you protect production bandwidth, how you handle the object store's lack of atomic rename for the jobs that will write there, and how you verify the copy is complete."

Solution Using DistCp snapshots + throttled bulk/incremental + the S3A magic committer

# 1. Freeze a source snapshot BEFORE the bulk run (exact delta later)
hdfs dfsadmin -allowSnapshot /warehouse
hdfs dfs -createSnapshot /warehouse s0

# 2. Throttled bulk copy — protect prod: cap per-mapper bandwidth,
#    run off-peak windows, dynamic strategy so big files don't stall
hadoop distcp -m 200 -bandwidth 15 -update -strategy dynamic \
  hdfs://nn/warehouse s3a://lake/warehouse

# 3. Second snapshot + exact incremental catch-up
hdfs dfs -createSnapshot /warehouse s1
hadoop distcp -m 100 -bandwidth 15 -update -diff s0 s1 \
  hdfs://nn/warehouse s3a://lake/warehouse

# 4. A FINAL micro-diff right before cutover (near-zero delta)
hdfs dfs -createSnapshot /warehouse s2
hadoop distcp -m 50 -bandwidth 25 -update -diff s1 s2 \
  hdfs://nn/warehouse s3a://lake/warehouse

# 5. All NEW writes on the lakehouse use the magic committer (no rename)
spark = (
    SparkSession.builder
    .config("spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a",
            "org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory")
    .config("spark.hadoop.fs.s3a.committer.name", "magic")
    .config("spark.hadoop.fs.s3a.committer.magic.enabled", "true")
    .config("spark.sql.sources.commitProtocolClass",
            "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
    .config("spark.sql.parquet.output.committer.class",
            "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")
    .getOrCreate()
)

# 6. Verify completeness — per-prefix file counts + byte totals
#    (HDFS CRC != S3 ETag, so compare counts/sizes, content-reconcile later)
hdfs dfs -count -q -h /warehouse/orders
hadoop fs -count -q -h s3a://lake/warehouse/orders

Step-by-step trace.

Step	Mechanism	Why it is safe
Snapshot s0	copy-on-write marker	exact delta base, no data duplicated
Bulk distcp	200 mappers @ 15 MB/s cap	throughput without starving prod
Diff s0→s1	snapshot delta	copies only files written during bulk
Micro-diff s1→s2	tiny delta at cutover	near-zero read-freeze window
Magic committer	multipart completion	writes never rely on rename
Count/size verify	per-prefix totals	catches gross copy gaps

After the bulk run copies 3 PB over ~three days at a throttled ~3 GB/s, the first snapshot diff picks up the ~20 TB written during those three days, and a final micro-diff at the cutover window copies the last few hundred GB in under an hour. Production never sees a bandwidth cliff because each mapper is capped. Every new job on the lakehouse commits via multipart completion, so a failed task never leaves partial output. Counts and sizes match per table prefix, and the critical tables get the content-level reconciliation from Section 5.

Output:

Metric	Value
Bulk copy	3.0 PB in ~62 h @ ~3 GB/s throttled
Incremental delta	~20 TB via snapshot -diff
Cutover micro-diff	< 1 TB, < 1 h read-freeze
Commit safety	multipart completion (no rename)
Verification	per-prefix counts + sizes; content-reconcile criticals
Production impact	bounded by per-mapper bandwidth cap

Why this works — concept by concept:

• HDFS snapshots + DistCp -diff — snapshots are copy-on-write metadata markers, so freezing s0 costs nothing and makes the later delta exact. -diff copies precisely the files that changed between two snapshots, turning a multi-day catch-up into a minutes-long delta instead of a full re-scan.
• Per-mapper bandwidth throttling — -bandwidth caps each mapper, so aggregate throughput is mappers × cap. This is how you copy at multi-GB/s without starving the production traffic that shares the link — the difference between a migration and an outage.
• Magic committer (multipart completion) — object stores have no atomic rename, so the classic FileOutputCommitter is unsafe. The magic committer uploads task output as incomplete multipart uploads and completes them at job commit — an atomic-per-file commit that is O(1) metadata, not an O(bytes) copy.
• Counts/sizes over cross-FS checksums — HDFS CRC32C and S3 ETag are different algorithms, so a checksum comparison across filesystems produces false mismatches. Counts and byte totals catch gross gaps; content reconciliation catches the subtle ones for tables that matter.
• Cost — the copy is a one-time O(bytes) transfer plus cheap snapshot deltas; the ongoing win is O(0) idle-compute spend once decommissioned, minus the 3× replication HDFS charged. The bulk transfer is the only large cost, and it buys the permanent storage/compute decoupling.

Data Processing
Topic — data-processing
Data-processing problems on distributed file copy and layout

Practice →

ETL Topic — etl ETL problems on incremental and idempotent loads

Practice →

---

3. Hive to Iceberg — the table-format migration

Hive to Iceberg swaps directory-listing for a snapshot log — and you adopt your existing Parquet without rewriting a single data file

The mental model in one line: migrating from the Hive table format to Iceberg replaces "a table is a directory whose subfolders are partitions, resolved by listing the filesystem and the metastore" with "a table is a metadata tree — a metadata.json pointing at a manifest list, pointing at manifests, pointing at data files — committed by atomically swapping one catalog pointer," and the critical property for a migration is that Iceberg can adopt your existing Parquet files in place via the snapshot, migrate, and add_files procedures, so you get snapshots, ACID commits, hidden partitioning, schema evolution, and time-travel rollback without a second petabyte-scale data rewrite. Every senior lakehouse migration hinges on this: the table format change is metadata work, not a data copy.

The four axes for the table-format migration.

• Adoption method. snapshot creates a new, independent Iceberg table that references the existing data files read-only — the Hive table is untouched, so it is the safe shadow you validate against. migrate converts the Hive table to Iceberg in place (the original is replaced; a backup table is kept). add_files imports files from a Hive table (or a path) into an existing Iceberg table. Choose snapshot for validation, migrate for the final cutover.
• Catalog choice. Iceberg needs a catalog that stores the current-metadata pointer per table: AWS Glue, a REST catalog (the open standard), Project Nessie (git-like branching), or a Hive-metastore-backed Iceberg catalog (reuse the existing HMS). The catalog is what makes commit atomic — it swaps one pointer.
• Partitioning & evolution. Hive partitioning is physical (the partition column literally names a directory) and cannot change without rewriting. Iceberg partitioning is hidden — a transform (days(ts), bucket(16, id)) recorded in metadata — so queries do not hard-code partition columns, and the partition spec can evolve without rewriting old data. Schema evolution is by column id, so a rename or reorder never corrupts old files.
• Maintenance. Iceberg tables need housekeeping the Hive format never had: rewrite_data_files (compact small files into large ones — critical after the HDFS→object-store move), expire_snapshots (drop old snapshots so metadata and orphaned files do not accumulate), and rewrite_manifests (keep the manifest tree balanced).

Iceberg table anatomy.

• metadata.json. The table root: current schema, partition spec(s), snapshot list, and the pointer to the current snapshot. A commit writes a new metadata.json and atomically points the catalog at it.
• Manifest list. One per snapshot — lists the manifest files that make up that snapshot, with partition-range summaries for pruning.
• Manifests. Each lists data files with per-file stats (row counts, column bounds, null counts) used to skip files at query time.
• Data files. Ordinary Parquet (or ORC/Avro) — the same files you already have. Adoption points the manifests at them; it does not rewrite them.

Adoption procedures — the migration verbs.

• snapshot. CALL catalog.system.snapshot('db.hive_orders', 'db.ice_orders') — creates ice_orders referencing hive_orders's files; the Hive table keeps working; writes to the snapshot do not affect the source. The validation table.
• migrate. CALL catalog.system.migrate('db.orders') — replaces db.orders with an Iceberg table over the same files and keeps db.orders_BACKUP_. The cutover verb.
• add_files. CALL catalog.system.add_files(table => 'db.ice_orders', source_table => 'db.hive_orders') — imports files from a Hive table/partition into an existing Iceberg table; used to top up a snapshot table or to merge multiple sources.

Common interview probes on the table-format migration.

• "Do you rewrite all the Parquet to move to Iceberg?" — no; snapshot/migrate/add_files adopt existing files.
• "snapshot vs migrate?" — snapshot = independent shadow (source untouched); migrate = in-place with a backup.
• "How does Iceberg commit atomically on S3?" — swaps one catalog metadata pointer; no rename of data files.
• "What must you run that Hive never needed?" — compaction (rewrite_data_files) and expire_snapshots.

Worked example — snapshot-then-validate (source stays untouched)

Detailed explanation. The safest first move is snapshot: it creates an independent Iceberg table over the copied Parquet, leaving the Hive table fully operational as the thing you reconcile against. You validate the Iceberg table's row counts and aggregates against the live Hive table, and only later do the in-place migrate. Walk through the snapshot and validation.

• Source. db.orders — a Hive table over s3a://lake/warehouse/orders (already copied).
• Shadow. glue.db.orders_ice — Iceberg, references the same files, independent.
• Validate. Compare counts and money sums between Hive and Iceberg.

Question. Create the Iceberg shadow via snapshot and write the validation queries that prove parity.

Input.

Object	Purpose
db.orders (Hive)	live source, untouched
glue.db.orders_ice	Iceberg shadow over same files
snapshot procedure	adopt files, no rewrite
parity queries	count + sum(amount)

Code.

-- 1. Create an INDEPENDENT Iceberg table over the existing Parquet.
--    The Hive table db.orders is NOT modified and keeps serving prod.
CALL glue.system.snapshot(
    source_table => 'db.orders',
    table        => 'glue.db.orders_ice'
);

-- 2. Sanity: the Iceberg table has a snapshot and references the same files
SELECT snapshot_id, committed_at, summary['total-records'] AS records
FROM   glue.db.orders_ice.snapshots
ORDER  BY committed_at DESC;

-- 3. Parity — row count (Hive engine vs Iceberg)
SELECT
    (SELECT count(*) FROM db.orders)            AS hive_rows,
    (SELECT count(*) FROM glue.db.orders_ice)   AS ice_rows;

-- 4. Parity — exact money sum + per-partition counts (the checks that
--    actually catch a bad adoption)
SELECT order_date,
       count(*)          AS ice_cnt,
       sum(amount_cents) AS ice_sum
FROM   glue.db.orders_ice
GROUP  BY order_date
ORDER  BY order_date;

# 5. Programmatic parity gate — fail loudly if anything drifts
def assert_parity(spark):
    hive = spark.sql("SELECT count(*) c, sum(amount_cents) s FROM db.orders").first()
    ice  = spark.sql("SELECT count(*) c, sum(amount_cents) s FROM glue.db.orders_ice").first()
    assert hive.c == ice.c, f"row drift: hive={hive.c} ice={ice.c}"
    assert hive.s == ice.s, f"sum drift:  hive={hive.s} ice={ice.s}"
    print(f"parity OK: {ice.c:,} rows, sum={ice.s:,}")

Step-by-step explanation.

1. CALL glue.system.snapshot(...) reads the Hive table's file list and metadata and writes an Iceberg metadata.json + manifests that point at the existing Parquet files. No data file is copied or rewritten; the operation is metadata-only and completes in seconds-to-minutes even for a huge table.
2. Because snapshot produces an independent table, db.orders (Hive) is completely untouched and keeps serving production. This is what makes it the safe validation baseline — you are comparing the new format against a still-live source, not against a copy of itself.
3. The snapshots metadata table confirms Iceberg recorded a snapshot and shows total-records, a first cheap sanity check that the manifest counts line up with expectations.
4. The parity queries are the real gate: an exact row count and an exact sum(amount_cents) (integer cents, so the sum is exact — never sum floats for a parity check) and per-partition counts. Per-partition counts localise any drift to a specific order_date, which is how you debug an adoption that missed a partition.
5. The programmatic assert_parity turns the check into a CI gate that fails loudly. During the wave, this runs on every dual-run so a regression can never slip past on a single lucky comparison.

Output.

Check	Hive	Iceberg	Status
Row count	4,812,004,331	4,812,004,331	match
sum(amount_cents)	918,224,551,900	918,224,551,900	match
Partitions present	1,096	1,096	match
Source table state	live, unmodified	independent shadow	safe

Rule of thumb. Start with snapshot, not migrate. It adopts the existing Parquet into an independent Iceberg table, leaves the Hive source live as your reconciliation baseline, and lets you validate row counts, exact integer sums, and per-partition counts before you ever touch the original.

Worked example — the in-place migrate cutover with a backup

Detailed explanation. Once the snapshot table has reconciled clean, the final table-format cutover is migrate: it converts the Hive table to Iceberg in place — same name, same files — and keeps a _BACKUP_ table so you can roll back. After migrate, the compute jobs read the same table name but now get Iceberg semantics. Walk through the migrate and the rollback path.

• Migrate. CALL catalog.system.migrate('db.orders') — db.orders becomes Iceberg; db.orders_BACKUP_ is the Hive original.
• Verify. Row count + sum against the pre-migrate snapshot.
• Rollback. Drop the Iceberg table, rename the backup back — the Hive table is intact.

Question. Perform the in-place migrate, verify it, and show the rollback if verification fails.

Input.

Component	Value
Target	db.orders (Hive → Iceberg in place)
Backup	db.orders_BACKUP_ (auto-kept)
Verify	count + sum vs pre-migrate
Rollback	drop Iceberg, rename backup back

Code.

-- 1. Capture the pre-migrate baseline from the validated shadow
--    (or from the live Hive table right before migrate)
CREATE TEMP VIEW pre_migrate AS
SELECT count(*) AS c, sum(amount_cents) AS s FROM db.orders;

-- 2. IN-PLACE migrate — db.orders becomes Iceberg over the SAME files.
--    Iceberg keeps db.orders_BACKUP_ (the original Hive table).
CALL glue.system.migrate(table => 'db.orders');

-- 3. Verify the migrated Iceberg table matches the baseline
SELECT p.c AS pre_rows, m.c AS post_rows,
       p.s AS pre_sum,  m.s AS post_sum,
       (p.c = m.c AND p.s = m.s) AS ok
FROM   pre_migrate p
CROSS  JOIN (SELECT count(*) c, sum(amount_cents) s FROM db.orders) m;

-- 4. Confirm it is now Iceberg (has a snapshot history)
SELECT count(*) AS snapshots FROM glue.db.orders.snapshots;

-- 5. ROLLBACK path — only if verification fails.
--    The Hive original is intact in the backup table.
DROP TABLE glue.db.orders;                       -- remove the Iceberg table
ALTER TABLE db.orders_BACKUP_ RENAME TO db.orders;   -- restore Hive original
-- Data files were never rewritten, so the original is byte-identical.

Step-by-step explanation.

1. Step 1 captures the pre-migrate baseline — an exact count and integer sum — so the post-migrate verification has something to compare against. In a wave, this baseline is the value the snapshot table already reconciled to.
2. CALL glue.system.migrate('db.orders') rewrites the table pointer, not the data. It builds Iceberg metadata over the existing files, registers db.orders as Iceberg in the catalog, and renames the old Hive definition to db.orders_BACKUP_. Downstream queries that reference db.orders now transparently get Iceberg.
3. Step 3's verification is the same exact count-and-sum parity, now comparing pre- and post-migrate. Because migrate does not touch data files, this must match exactly; any drift means the adoption misread the file list and you roll back.
4. Step 4 confirms the table is genuinely Iceberg by checking it has a snapshot history — a Hive table has none. This distinguishes "migrate succeeded" from "migrate silently no-op'd."
5. The rollback (Step 5) is trivial precisely because migrate never rewrote data: drop the Iceberg table (removing only its metadata) and rename the backup back to the original name. The Parquet files are byte-identical throughout, so the Hive table returns exactly as it was.

Output.

Stage	db.orders format	Backup present	Rollback cost
Before migrate	Hive	—	n/a
After migrate	Iceberg (same files)	db.orders_BACKUP_	metadata-only
Verify	count+sum match	yes	—
Rollback (if needed)	Hive restored	consumed	seconds

Rule of thumb. Do the in-place migrate only after the snapshot shadow has reconciled clean, verify with the same exact count-and-sum parity, and lean on the auto-kept _BACKUP_ table — because migrate never rewrites data files, rollback is a metadata rename, not a restore.

Worked example — partition-spec evolution and post-move compaction

Detailed explanation. Two things you can do on Iceberg that Hive made impossible or painful: change the partitioning without rewriting old data (hidden partitioning + partition-spec evolution), and compact the small files that the HDFS→object-store move inevitably left behind (every small object is a separate GET). Walk through evolving the spec and running compaction as routine maintenance.

• Hidden partitioning. Partition by days(order_ts) — queries filter on order_ts and Iceberg prunes automatically; the partition column is not a physical directory name.
• Spec evolution. Switch from days to hours for new data without rewriting the years of old daily-partitioned files.
• Compaction. rewrite_data_files bins many small Parquet files into fewer large ones; expire_snapshots reclaims the orphaned small files.

Question. Set hidden partitioning, evolve the spec, and run the compaction + snapshot-expiry maintenance.

Input.

Operation	Iceberg mechanism
Hidden partition	PARTITIONED BY days(order_ts)
Evolve spec	ALTER TABLE ... ADD PARTITION FIELD hours(order_ts)
Compact small files	rewrite_data_files
Reclaim storage	expire_snapshots

Code.

-- 1. Hidden partitioning: partition by a TRANSFORM of the timestamp.
--    Queries filter on order_ts; Iceberg prunes without a partition column.
ALTER TABLE glue.db.orders
    ADD PARTITION FIELD days(order_ts);

-- 2. Evolve the spec for NEW data (finer granularity) WITHOUT rewriting
--    the historical daily-partitioned files. Old + new specs coexist.
ALTER TABLE glue.db.orders
    ADD PARTITION FIELD hours(order_ts);

-- 3. Compact small files (the small-object problem after HDFS→S3).
--    Bin-pack many small Parquet files into ~512 MB targets.
CALL glue.system.rewrite_data_files(
    table   => 'db.orders',
    options => map('target-file-size-bytes', '536870912',
                   'min-input-files',        '5')
);

-- 4. Expire old snapshots so metadata + orphaned small files are reclaimed
CALL glue.system.expire_snapshots(
    table            => 'db.orders',
    older_than       => TIMESTAMP '2026-08-11 00:00:00',
    retain_last      => 5
);

-- 5. Rebalance the manifest tree after large rewrites
CALL glue.system.rewrite_manifests(table => 'db.orders');

Step-by-step explanation.

1. Adding days(order_ts) as a partition field is hidden partitioning: the physical layout is bucketed by day, but queries just say WHERE order_ts >= ... and Iceberg maps that to the right partitions using the recorded transform. There is no dt=2026-08-17 directory a query must know about, so you cannot get the classic Hive "forgot the partition predicate, full-scanned the table" bug.
2. Adding hours(order_ts) evolves the spec. Iceberg keeps the old daily spec for existing files and applies the new hourly spec to new writes — the two coexist in one table. Hive would have forced a full rewrite to repartition; Iceberg makes it a metadata change.
3. rewrite_data_files addresses the small-file problem the object-store move created: bin-packing many sub-optimal Parquet files into ~512 MB targets so scans issue fewer, larger reads. min-input-files=5 avoids rewriting partitions that are already tidy.
4. expire_snapshots is the reclaim step: compaction produces new files and leaves the old small ones referenced only by historical snapshots. Expiring snapshots older than the retention window (keeping the last 5) lets Iceberg delete those now-orphaned files and shrink metadata.
5. rewrite_manifests rebalances the manifest tree after a large rewrite so that partition pruning stays fast. Together, these three procedures are the routine maintenance an Iceberg table needs and a Hive table never had — and skipping them is why some migrations report "Iceberg is slower," when the real problem is uncompacted small files.

Output.

Maintenance	Before	After
Files per day-partition	~1,800 small	~40 (~512 MB)
Partition granularity	days only	days (old) + hours (new)
Query pruning	manual predicate on old Hive	hidden, automatic
Storage after expiry	orphans retained	orphans reclaimed

Rule of thumb. Use hidden partitioning so queries cannot forget the partition predicate, evolve the spec instead of rewriting to repartition, and schedule rewrite_data_files + expire_snapshots as routine maintenance — the small-file compaction is what makes the post-move Iceberg table faster than the Hive original, not slower.

Senior interview question on the table-format migration

A senior interviewer might ask: "You've copied 800 Hive tables' Parquet to S3. Now convert them to Iceberg with zero data rewrite, validate each one against the live Hive table before cutting downstream queries over, keep a rollback, and explain the ongoing maintenance the Iceberg tables will need that the Hive tables never did. Walk me through the procedures and the validation."

Solution Using snapshot-validate-then-migrate with compaction and snapshot expiry

-- 1. SHADOW: adopt existing Parquet into an independent Iceberg table.
--    Hive source stays live -> it is the reconciliation baseline.
CALL glue.system.snapshot(
    source_table => 'db.orders',
    table        => 'glue.db.orders_ice'
);

-- 2. VALIDATE: exact parity (count + integer sum + per-partition counts)
SELECT
    (SELECT count(*)            FROM db.orders)          AS hive_rows,
    (SELECT count(*)            FROM glue.db.orders_ice) AS ice_rows,
    (SELECT sum(amount_cents)   FROM db.orders)          AS hive_sum,
    (SELECT sum(amount_cents)   FROM glue.db.orders_ice) AS ice_sum;

-- 3. CUTOVER: only after N clean reconciles, migrate IN PLACE with backup
CALL glue.system.migrate(table => 'db.orders');   -- keeps db.orders_BACKUP_

-- 4. MAINTENANCE that Hive never needed:
--    (a) compact the small files the HDFS->S3 move produced
CALL glue.system.rewrite_data_files(
    table   => 'db.orders',
    options => map('target-file-size-bytes','536870912','min-input-files','5'));

--    (b) expire old snapshots to reclaim orphaned files + trim metadata
CALL glue.system.expire_snapshots(
    table => 'db.orders', older_than => current_timestamp - INTERVAL '7' DAY,
    retain_last => 5);

--    (c) keep the manifest tree balanced for fast pruning
CALL glue.system.rewrite_manifests(table => 'db.orders');

# 5. Wave driver — snapshot+validate all 800 tables; migrate only the clean ones
def migrate_wave(spark, tables):
    clean, dirty = [], []
    for t in tables:
        spark.sql(f"CALL glue.system.snapshot('db.{t}', 'glue.db.{t}_ice')")
        h = spark.sql(f"SELECT count(*) c, sum(amount_cents) s FROM db.{t}").first()
        i = spark.sql(f"SELECT count(*) c, sum(amount_cents) s FROM glue.db.{t}_ice").first()
        (clean if (h.c == i.c and h.s == i.s) else dirty).append(t)
    for t in clean:                     # cut over only what reconciled exactly
        spark.sql(f"CALL glue.system.migrate('db.{t}')")
    return {"migrated": clean, "held_for_investigation": dirty}

Step-by-step trace.

Phase	Procedure	Data rewritten?	Rollback
Shadow	snapshot	no (adopts Parquet)	drop shadow
Validate	count + sum parity	no	n/a
Cutover	migrate (in place)	no (same files)	rename BACKUP
Compact	rewrite_data_files	yes (small→large)	prior snapshot
Reclaim	expire_snapshots	deletes orphans	within retention

After the wave, all 800 tables that reconciled exactly are Iceberg over their original Parquet — no data file was rewritten during adoption or cutover — and each keeps a _BACKUP_ for the rollback window. Compaction then bin-packs the small files the object-store move produced, expire_snapshots reclaims the orphans, and rewrite_manifests keeps pruning fast. Any table whose count or sum drifted is held out of cutover and investigated rather than migrated blind.

Output:

Metric	Value
Tables adopted with zero rewrite	800 (snapshot/migrate)
Cutover gate	exact count + integer-sum parity
Rollback	rename BACKUP (metadata-only)
Post-move compaction	~1,800 → ~40 files/partition
New maintenance	rewrite_data_files, expire_snapshots, rewrite_manifests
Held for investigation	any table with count/sum drift

Why this works — concept by concept:

• snapshot as an independent shadow — adoption is metadata-only: Iceberg writes manifests pointing at the existing Parquet, so no petabyte is rewritten and the Hive source stays live as the reconciliation baseline. This is what makes validate-before-cutover possible.
• Exact integer parity gate — comparing count(*) and sum(amount_cents) (integer cents, never floats) plus per-partition counts is an exact invariant; a single mismatched partition fails the gate and holds the table out of cutover. Correctness is asserted, not hoped.
• migrate keeps a backup, rewrites no data — in-place cutover swaps the catalog pointer and renames the Hive original to _BACKUP_. Because data files are untouched, rollback is a rename, not a restore — the cheapest possible undo.
• Compaction + snapshot expiry — the maintenance Hive never had. rewrite_data_files fixes the small-object problem the HDFS→S3 move created; expire_snapshots reclaims orphaned files; rewrite_manifests keeps pruning fast. Skipping these is why a naive migration reports Iceberg as "slower."
• Cost — adoption is O(files) metadata, not O(bytes) data; the only O(bytes) work is compaction, which you run once post-move and then incrementally. Compared to a reload-into-Iceberg approach (a second full petabyte rewrite), snapshot/migrate is orders of magnitude cheaper and is the only tractable path at scale.

Data Transformation
Topic — data-transformation
Data-transformation problems on table formats and schema evolution

Practice →

Optimization Topic — optimization Optimization problems on partitioning and file compaction

Practice →

---

4. Job rewrites — MapReduce / HiveQL / Pig → Spark

A Spark migration is mostly a HiveQL-to-Spark-SQL port — but parity lives and dies in the semantic edge cases you pin with a golden-output diff

The mental model in one line: the job rewrite layer reclassifies every job by how much it actually changes — HiveQL is mostly Spark-SQL-compatible so most Hive jobs are a light port, MapReduce and Pig are full rewrites into Spark, and Oozie coordinators become Airflow DAGs — and the entire correctness risk is concentrated in a short list of semantic differences (implicit casts, NULL versus empty string, decimal precision, timezone handling, LATERAL VIEW versus explode) that you catch not by reading the code but by running the old and new jobs on the same input and diffing the output byte-for-byte. Every senior migration owns a golden-output test harness; the engineers who "port and eyeball" are the ones who ship a decimal-rounding drift into a finance table.

The four axes for the job-rewrite migration.

• Rewrite depth per job class. HiveQL → Spark SQL is usually a near-copy (same dialect family), so the bulk of jobs are light. MapReduce (Java mapper/reducer) and Pig Latin are full rewrites into Spark DataFrame/SQL. Oozie XML → Airflow Python is a scheduling rewrite. Scope the project by counting jobs in each class, because the cost is dominated by the MapReduce/Pig long tail, not the HiveQL bulk.
• Semantic parity. The traps: HiveQL treats some implicit type coercions differently from Spark; NULL versus empty-string handling in LOAD/split; decimal precision and rounding; from_unixtime/timezone defaults; LATERAL VIEW explode becomes explode(); reserved-word and identifier-quoting differences; CLUSTER BY/DISTRIBUTE BY semantics. Each is a place output can silently differ.
• Compute decoupling. YARN → Spark on Kubernetes / EMR / Dataproc / serverless. This is where autoscaling and scale-to-zero land, but it also changes shuffle behaviour, memory config, and dynamic allocation — so a job that was tuned for a fixed YARN queue often needs re-tuning, and "same logic, different runtime" can still change performance (not correctness).
• Validation. The golden-output diff: run the legacy job and the rewritten job on an identical frozen input, then compare outputs with an order-independent, type-aware diff (sorted, with exact money columns and tolerance only where genuinely floating-point). This is the gate that lets you cut a job over.

The job inventory — classify before you rewrite.

• HiveQL scripts. Count them; most port with minor edits. Run each through Spark SQL and diff.
• MapReduce jobs. Full rewrites; the expensive long tail. Reimplement as Spark DataFrame transformations, not a line-by-line Java translation.
• Pig scripts. Full rewrites into Spark SQL/DataFrame; Pig's dataflow maps cleanly to a chain of transforms.
• Oozie workflows. Rewrite scheduling and dependencies as Airflow DAGs; the data logic is unchanged, only the orchestration.

The semantic-trap checklist — where parity breaks.

• Implicit casts. Hive and Spark disagree on some string↔number coercions; make casts explicit.
• NULL vs empty string. Hive's text SerDe can turn '' into NULL or vice versa; pin it with explicit handling.
• Decimal precision. DECIMAL scale/rounding differs; set precision explicitly and never sum as double for money.
• Timezone. from_unixtime / to_utc_timestamp defaults differ; set spark.sql.session.timeZone and be explicit.
• LATERAL VIEW / explode. HiveQL LATERAL VIEW explode(col) becomes Spark's explode(); watch outer-explode for empty arrays.

Common interview probes on the job-rewrite migration.

• "Is HiveQL compatible with Spark SQL?" — mostly, but pin the semantic traps; prove parity with a diff.
• "How do you migrate a MapReduce job?" — full rewrite as Spark DataFrame logic, not a Java line-port.
• "How do you know the rewrite is correct?" — golden-output diff on identical frozen input.
• "What changes moving off YARN?" — autoscaling and scale-to-zero, plus shuffle/memory re-tuning.

Worked example — porting a HiveQL job to Spark SQL with a golden-output diff

Detailed explanation. The canonical light rewrite: a HiveQL aggregation that runs on Hive-on-Tez is ported to Spark SQL. The SQL is nearly identical, but you pin the decimal and timezone semantics and then prove parity by running both on the same frozen input and diffing. Walk through the port and the diff harness.

• Legacy. A HiveQL daily revenue rollup on Hive-on-Tez.
• Port. The same SQL as Spark SQL, with explicit decimal cast and session timezone set to UTC.
• Prove. Run both on a frozen input partition; diff sorted output with exact money comparison.

Question. Port the HiveQL rollup to Spark SQL and write the golden-output diff that gates the cutover.

Input.

Aspect	Legacy (Hive)	New (Spark)
Engine	Hive-on-Tez	Spark SQL
Decimal	implicit	explicit CAST(... AS DECIMAL(18,2))
Timezone	server default	spark.sql.session.timeZone=UTC
Validation	none	golden-output diff

Code.

-- Legacy HiveQL (runs on Hive-on-Tez)
INSERT OVERWRITE TABLE mart.daily_revenue PARTITION (dt)
SELECT region,
       SUM(amount) AS revenue,          -- implicit decimal; server tz
       COUNT(*)    AS orders,
       dt
FROM   db.orders
WHERE  dt = '2026-08-17'
GROUP  BY region, dt;

# Rewritten as Spark SQL — semantics pinned explicitly
from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("daily-revenue")
         .config("spark.sql.session.timeZone", "UTC")          # pin tz
         .getOrCreate())

spark.sql("""
    INSERT OVERWRITE TABLE mart.daily_revenue PARTITION (dt)
    SELECT region,
           CAST(SUM(CAST(amount AS DECIMAL(18,2))) AS DECIMAL(18,2)) AS revenue,
           COUNT(*) AS orders,
           dt
    FROM   glue.db.orders
    WHERE  dt = '2026-08-17'
    GROUP  BY region, dt
""")

# Golden-output diff — run both on the SAME frozen input, compare exactly
def golden_diff(spark, hive_out: str, spark_out: str, keys, money_cols):
    h = spark.read.parquet(hive_out).orderBy(*keys)
    s = spark.read.parquet(spark_out).orderBy(*keys)
    # 1. same row count
    assert h.count() == s.count(), "row count differs"
    # 2. exact join on keys; flag any column that differs
    j = h.alias("h").join(s.alias("s"), on=keys, how="full_outer")
    diffs = j.where(
        " OR ".join(f"NOT (h.{c} <=> s.{c})"          # <=> is null-safe equals
                    for c in money_cols + ["orders"])
    )
    n = diffs.count()
    assert n == 0, f"{n} rows differ between Hive and Spark output"
    print("golden diff: PASS (byte-parity on keys + measures)")

Step-by-step explanation.

1. The legacy HiveQL sums amount with implicit decimal handling and the server's default timezone — both invisible until they differ from Spark. Copying it verbatim to Spark is the trap: it looks identical and reads correct, but the coercions can round differently.
2. The Spark rewrite pins the two traps that matter here: spark.sql.session.timeZone=UTC fixes any date-bucketing that depends on timezone, and the explicit CAST(... AS DECIMAL(18,2)) fixes the summation precision so money never rounds differently from Hive.
3. The SQL body is otherwise unchanged — this is the point that "HiveQL is mostly Spark-SQL-compatible." The rewrite effort per HiveQL job is small; the validation effort is where the real work is.
4. The golden_diff harness runs both jobs on the same frozen input partition and compares outputs. It sorts by keys (Spark output order is non-deterministic), checks row counts, then full-outer-joins and flags any measure that differs using <=> (null-safe equality) so NULL-vs-NULL is treated as equal and NULL-vs-value as a difference.
5. The assertion is the cutover gate: zero differing rows means byte-parity on the keys and measures, and only then does the Spark job replace the Hive job for that table. A single differing row blocks cutover and points you straight at the offending region/measure.

Output.

Check	Result
Row count (Hive vs Spark)	equal
revenue per region	exact match (DECIMAL(18,2))
orders per region	exact match
Timezone-sensitive dt bucketing	identical (UTC pinned)
Cutover gate	PASS → Spark job replaces Hive job

Rule of thumb. Port HiveQL to Spark SQL almost verbatim, but pin the semantic traps explicitly — session timezone and decimal casts first — and never cut a job over on a code read. The golden-output diff on frozen input, with null-safe exact comparison of money columns, is the only gate that catches a silent rounding drift.

Worked example — rewriting a MapReduce job as a Spark DataFrame job

Detailed explanation. A Java MapReduce job — the expensive long tail — is reimplemented as Spark DataFrame logic, not translated line-by-line. The mapper/reducer's intent (parse, key, aggregate) maps to select/groupBy/agg. Walk through converting a classic word-count-style sessionization mapper/reducer to Spark.

• Legacy. A MapReduce job: mapper parses log lines and emits (user_id, event), reducer counts events per user.
• Rewrite. Read the logs as a DataFrame, parse, group by user, aggregate — a handful of transforms.
• Prove. Golden-output diff against the MapReduce output on the same input.

Question. Reimplement the mapper/reducer as a Spark DataFrame job and note what replaces each MapReduce stage.

Input.

MapReduce stage	Spark equivalent
InputFormat / record reader	spark.read (text/parquet)
map() emit (k, v)	select / withColumn
shuffle + sort by key	groupBy
reduce() aggregate	agg
OutputFormat	write

Code.

# Legacy MapReduce (Java): mapper emits (user_id, 1) per event line;
# reducer sums per user. Reimplemented as Spark DataFrame logic:
from pyspark.sql import functions as F

events = (
    spark.read.parquet("s3a://lake/warehouse/events")   # was: InputFormat
    .where("dt = '2026-08-17'")
)

per_user = (
    events
    .select(                                            # was: map() emit
        F.col("user_id"),
        F.col("event_type"),
    )
    .groupBy("user_id")                                 # was: shuffle by key
    .agg(                                               # was: reduce()
        F.count("*").alias("event_count"),
        F.countDistinct("event_type").alias("distinct_events"),
    )
)

(
    per_user
    .repartition(200)                                   # control output files
    .write.mode("overwrite")                            # was: OutputFormat
    .parquet("s3a://lake/warehouse/user_event_counts/dt=2026-08-17")
)

# Prove parity against the MapReduce output on the SAME input partition
mr  = spark.read.parquet("s3a://legacy/user_event_counts/dt=2026-08-17")
new = spark.read.parquet("s3a://lake/warehouse/user_event_counts/dt=2026-08-17")

assert mr.count() == new.count(), "row count differs"
delta = (mr.alias("a").join(new.alias("b"), "user_id", "full_outer")
           .where("NOT (a.event_count <=> b.event_count) OR "
                  "NOT (a.distinct_events <=> b.distinct_events)"))
assert delta.count() == 0, "per-user counts differ between MR and Spark"
print("MapReduce → Spark parity: PASS")

Step-by-step explanation.

1. The MapReduce InputFormat/record-reader becomes spark.read.parquet(...) — Spark handles splitting and reading. You do not reimplement the record reader; you declare the source.
2. The mapper's "emit (user_id, event)" becomes a select of the columns you key and aggregate on. There is no explicit emit; the DataFrame is the intermediate representation.
3. The MapReduce shuffle-and-sort-by-key is exactly groupBy("user_id") — Spark's Catalyst planner inserts the shuffle. You express intent (group by user), not the mechanics (partitioner, comparator, combiner).
4. The reducer's aggregation becomes agg(count(...), countDistinct(...)). A countDistinct that in MapReduce needed a secondary-sort or an in-reducer set is a single built-in in Spark — this is why a reimplementation is smaller and clearer than a line-by-line port.
5. Parity is proven the same way as the HiveQL case: run both on the same frozen partition, count rows, full-outer-join on the key, and assert null-safe equality of every measure. A reimplementation must still pass a byte-parity gate before it replaces the MapReduce job.

Output.

MapReduce artifact	Lines of Java (approx)	Spark equivalent
Mapper + Reducer + Driver	~180	~12 lines DataFrame
Shuffle/sort/partitioner config	manual	Catalyst-inserted
Distinct-count logic	secondary sort	countDistinct built-in
Parity gate	none historically	golden-output diff

Rule of thumb. Reimplement MapReduce jobs as Spark DataFrame intent — read / select / groupBy / agg / write — rather than translating Java stage-by-stage. Let Catalyst own the shuffle, use built-in aggregates, and still gate the cutover on a golden-output diff.

Worked example — Oozie coordinator to an Airflow DAG

Detailed explanation. The orchestration layer is the lowest-risk rewrite because the data logic does not change — only the scheduling and dependency wiring. An Oozie coordinator (XML: schedule + workflow of actions with data dependencies) becomes an Airflow DAG (Python: schedule + tasks with dependencies). Walk through translating a daily Oozie coordinator into an Airflow DAG.

• Legacy. Oozie coordinator: runs daily, waits for an input _SUCCESS marker, then runs a Hive action, then a shell action.
• Rewrite. Airflow DAG: @daily schedule, a sensor for the input, a Spark task, a downstream task, wired with >>.
• Validate. Same-schedule dry run; confirm task order and data dependency match.

Question. Translate the Oozie coordinator into an Airflow DAG preserving the schedule and the data dependency.

Input.

Oozie concept	Airflow equivalent
coordinator frequency	DAG schedule (@daily)
dataset / done-flag	sensor (S3KeySensor)
workflow action (Hive)	task (SparkSubmit / SQL)
action ok-to → next	dependency (task_a >> task_b)

Code.

# Oozie coordinator (XML) had: frequency=daily, an input-event waiting on
# a _SUCCESS marker, then a Hive action, then a shell action.
# Airflow equivalent:
from airflow import DAG
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="daily_revenue",
    schedule="@daily",                       # was: coordinator frequency
    start_date=datetime(2026, 8, 1),
    catchup=False,
) as dag:

    wait_for_input = S3KeySensor(            # was: input-event / done-flag
        task_id="wait_for_orders",
        bucket_key="s3://lake/warehouse/orders/dt={{ ds }}/_SUCCESS",
        poke_interval=300, timeout=6 * 3600,
    )

    revenue = SparkSubmitOperator(           # was: Hive action
        task_id="daily_revenue_rollup",
        application="/jobs/daily_revenue.py",
        application_args=["--dt", "{{ ds }}"],
    )

    notify = BashOperator(                    # was: shell action
        task_id="notify_downstream",
        bash_command="curl -X POST $HOOK -d 'daily_revenue ready for {{ ds }}'",
    )

    wait_for_input >> revenue >> notify       # was: action ok-to transitions

Step-by-step explanation.

1. The Oozie coordinator's frequency=daily becomes the DAG's schedule="@daily". Airflow's {{ ds }} template gives the logical date, the direct analogue of Oozie's ${coord:current(0)} nominal time — so the daily partition each run targets is preserved.
2. Oozie's input-event that waited on a dataset's _SUCCESS done-flag becomes an S3KeySensor polling for the _SUCCESS marker on the day's partition. The data dependency — "do not start until the input is ready" — is preserved exactly, just expressed as a sensor.
3. The Oozie Hive action becomes a SparkSubmitOperator running the rewritten Spark job (from the previous examples). The orchestration rewrite and the compute rewrite meet here: Airflow schedules the Spark job that replaced the Hive job.
4. The Oozie shell action becomes a BashOperator. Non-data actions (notifications, triggers) translate one-to-one.
5. Oozie's ok-to transitions between actions become Airflow's >> dependency operator: wait_for_input >> revenue >> notify. The DAG topology is the workflow graph; the schedule and every data dependency are preserved, so a same-schedule dry run should produce the same runs at the same times as Oozie did.

Output.

Oozie element	Airflow element	Preserved
frequency=daily	schedule="@daily"	schedule
input done-flag	S3KeySensor on _SUCCESS	data dependency
Hive action	SparkSubmitOperator	step (now Spark)
ok-to transitions	task_a >> task_b	topology

Rule of thumb. Translate Oozie to Airflow one concept at a time — frequency→schedule, done-flag→sensor, action→operator, ok-to→>> — and validate with a same-schedule dry run. The data logic is unchanged, so this layer is the safest of the four; the only risk is a mistranslated dependency, which the dry run surfaces.

Senior interview question on the job-rewrite migration

A senior interviewer might ask: "You have 600 jobs: 450 HiveQL scripts, 90 MapReduce jobs, 40 Pig scripts, and 20 Oozie coordinators, all feeding finance-critical tables. Walk me through how you classify and rewrite them onto Spark, the semantic traps that will silently break parity, how you prove each rewrite is correct before cutover, and what changes when you move off YARN."

Solution Using a classified rewrite with pinned semantics and a golden-output gate

# 1. CLASSIFY the inventory by rewrite depth (cost is in the MR/Pig tail)
inventory = {
    "hiveql":    450,   # light port: HiveQL -> Spark SQL, pin semantics
    "mapreduce":  90,   # full rewrite: Java -> Spark DataFrame intent
    "pig":        40,   # full rewrite: Pig dataflow -> Spark transforms
    "oozie":      20,   # scheduling rewrite: XML -> Airflow DAG
}

# 2. PIN the semantic traps globally so every job inherits safe defaults
spark = (SparkSession.builder.appName("migration")
    .config("spark.sql.session.timeZone", "UTC")                  # timezone
    .config("spark.sql.storeAssignmentPolicy", "ANSI")           # strict casts
    .config("spark.sql.ansi.enabled", "true")                    # no silent coercion
    .config("spark.sql.parquet.int96RebaseModeInRead", "CORRECTED")
    .getOrCreate())
# Per-job: explicit CAST(... AS DECIMAL(p,s)) for money; explode() for
# LATERAL VIEW; explicit NULL/'' handling in text parsing.

# 3. GOLDEN-OUTPUT GATE — every rewrite must byte-match legacy on frozen input
def cutover_gate(spark, legacy_path, new_path, keys, measures):
    a = spark.read.parquet(legacy_path)
    b = spark.read.parquet(new_path)
    if a.count() != b.count():
        return False, "row count differs"
    cond = " OR ".join(f"NOT (a.{m} <=> b.{m})" for m in measures)
    n = (a.alias("a").join(b.alias("b"), keys, "full_outer").where(cond)).count()
    return (n == 0), (f"{n} rows differ" if n else "byte-parity")

# 4. Only jobs whose gate returns True are cut over; the rest are held.

# 5. Compute decoupling — YARN queue -> Spark on Kubernetes with autoscale
spark.kubernetes.container.image: registry/spark:3.5
spark.dynamicAllocation.enabled: "true"
spark.dynamicAllocation.minExecutors: "0"      # scale to zero off-peak
spark.dynamicAllocation.maxExecutors: "400"    # burst at peak
spark.dynamicAllocation.shuffleTracking.enabled: "true"

Step-by-step trace.

Job class	Count	Rewrite depth	Parity method
HiveQL	450	light (SQL port)	golden diff
MapReduce	90	full (DataFrame)	golden diff
Pig	40	full (DataFrame)	golden diff
Oozie	20	scheduling only	same-schedule dry run
YARN → K8s	—	runtime swap	perf re-tune (not correctness)

After classification, the 450 HiveQL jobs port quickly but each still passes the golden-output gate; the 130 MapReduce/Pig jobs are the real effort and are reimplemented as Spark DataFrame logic; the 20 Oozie coordinators become Airflow DAGs validated by a same-schedule dry run. ANSI mode and a pinned UTC timezone make casts strict and dates deterministic, so the semantic traps surface as errors at rewrite time rather than silent drift at cutover. Moving to Spark-on-Kubernetes with dynamic allocation scales executors to zero off-peak — a performance and cost change, gated separately from correctness.

Output:

Metric	Value
Jobs classified	600 (450/90/40/20)
Correctness gate	golden-output diff on frozen input
Semantic traps	pinned via ANSI + UTC + explicit casts
Orchestration validation	same-schedule dry run
Compute	Spark on K8s, minExecutors=0
Cutover rule	only gate-passing jobs move; rest held

Why this works — concept by concept:

• Classify by rewrite depth — HiveQL is a light SQL port, MapReduce/Pig are full DataFrame reimplementations, Oozie is scheduling-only. Counting jobs per class scopes the real cost (the MR/Pig long tail) instead of assuming every job is equal.
• Pin semantics globally — ANSI mode makes implicit casts throw instead of silently coercing, and a fixed UTC session timezone makes date bucketing deterministic. The semantic traps become loud rewrite-time errors rather than quiet cutover-time drift.
• Golden-output diff — running legacy and rewritten jobs on the same frozen input and comparing with null-safe exact equality on measures is the only gate that proves parity. Code reads miss decimal-rounding and NULL-handling drift; a byte diff cannot.
• Orchestration is scheduling-only — Oozie→Airflow changes when and in what order jobs run, not what they compute, so a same-schedule dry run is sufficient validation and this layer is the safest.
• Cost — the rewrite is O(jobs) engineer-effort weighted toward the MR/Pig tail; the runtime swap to autoscaled Spark-on-K8s is where the ongoing compute cost drops (minExecutors=0 off-peak). Correctness and performance are gated separately so a re-tune never masquerades as a correctness regression.

Data Processing
Topic — data-processing
Data-processing problems on Spark rewrites and aggregation

Practice →

Optimization Topic — optimization Optimization problems on shuffle, partitioning, and Spark tuning

Practice →

---

5. Cutover, reconciliation & cluster decommission

Cluster decommission is the last step, not the first — you dual-run, reconcile until the numbers match, cut over table-by-table, and only then power the DataNodes off

The mental model in one line: the cutover phase runs the legacy Hadoop pipeline and the new lakehouse pipeline in parallel on the same inputs, reconciles their outputs with exact row counts, integer money sums, and per-partition checks until they match for N consecutive runs, cuts each downstream consumer over table-by-table with a warm-legacy rollback window, and only performs the irreversible cluster decommission — draining YARN, retiring DataNodes, deleting HDFS — after the last consumer has moved and every rollback window has closed, because that final step is both the source of the cost saving and the point of no return. Every senior migration is judged on this phase: the copy and the rewrite are recoverable mistakes; a premature decommission is not.

The four axes for the cutover.

• Dual-run. Both stacks produce every table for a period. The legacy output is the reference; the lakehouse output is the candidate. Running both costs double temporarily — which is exactly why the rollback windows are time-boxed and decommission is not delayed indefinitely.
• Reconciliation. The comparison that gates cutover: exact count(*), exact integer sum() of money columns (never float sums), per-partition counts, and for the highest-value tables a full column-level diff or a sampled row hash. A single clean run is not enough — you require N consecutive clean runs to rule out a flaky pass.
• Cutover strategy. Table-by-table within a wave, repointing each downstream consumer from the legacy table to the Iceberg table. Every cutover keeps the legacy output warm for a rollback window so a consumer that finds a problem days later can be flipped back in minutes.
• Decommission. The terminal, irreversible step: drain the YARN queues so no new work schedules, retire the NodeManagers, archive a final checksum manifest of the HDFS warehouse, delete the HDFS data, and power down the DataNodes. This is where the storage/compute-decoupling cost saving is finally realised rather than merely projected.

The reconciliation harness — what "the numbers match" means.

• Counts. count(*) per table and per partition — the cheapest, first-line check.
• Exact sums. Integer sums of money/quantity columns (store money as integer cents so the sum is exact); an equal count with an unequal sum means a value-level bug the count missed.
• Distinct keys. count(distinct pk) catches duplicate or dropped keys that a raw count can hide.
• Sampled row hash. For the critical tables, hash a deterministic sample of rows and compare — catches per-column drift that aggregate checks miss.

The decommission checklist — the point of no return.

• Drain. Stop scheduling new YARN work; let in-flight jobs finish.
• Confirm no consumers. Assert every downstream job now reads the Iceberg tables — no lingering HDFS path references.
• Archive. Write a final checksum/manifest of the HDFS warehouse to cold storage (audit + last-resort restore).
• Delete & power off. Delete HDFS data, retire DataNodes — the cost saving is realised here.

Common interview probes on the cutover.

• "How do you validate a migration without a big-bang?" — dual-run + reconcile counts/sums/partitions for N runs.
• "What do you check beyond row counts?" — exact integer sums, distinct keys, sampled row hashes.
• "How do you roll back after cutover?" — repoint the consumer at the warm legacy output within the rollback window.
• "When can you decommission?" — only after the last consumer moves and every rollback window closes.

Worked example — the reconciliation harness

Detailed explanation. The reconciliation harness is a job that reads the legacy output and the lakehouse output for a table, runs the tiered checks (counts → sums → distinct → sampled hash), and records a pass/fail per run. Cutover is gated on N consecutive passes. Walk through the harness.

• Inputs. Legacy Hive/HDFS output and Iceberg output for the same partition.
• Checks. Count, exact integer sum, distinct PK, sampled row hash — cheapest to most thorough.
• Gate. Record pass/fail; require N consecutive passes before cutover.

Question. Implement the tiered reconciliation and the N-consecutive-pass gate.

Input.

Tier	Check	Catches
1	count(*)	missing/extra rows
2	sum(amount_cents)	value drift at equal count
3	count(distinct pk)	dup/dropped keys
4	sampled row hash	per-column drift

Code.

from pyspark.sql import functions as F

def reconcile(spark, legacy: str, candidate: str, pk: str, money: str) -> dict:
    a = spark.read.format("parquet").load(legacy)          # Hadoop output
    b = spark.table(candidate)                             # Iceberg output

    # Tier 1–3: exact aggregate invariants
    ra = a.agg(F.count("*").alias("c"),
               F.sum(money).alias("s"),
               F.countDistinct(pk).alias("d")).first()
    rb = b.agg(F.count("*").alias("c"),
               F.sum(money).alias("s"),
               F.countDistinct(pk).alias("d")).first()

    checks = {
        "count":    ra.c == rb.c,
        "sum":      ra.s == rb.s,        # integer cents -> exact
        "distinct": ra.d == rb.d,
    }

    # Tier 4: sampled deterministic row hash (critical tables only)
    def sampled_hash(df):
        return (df.where(F.crc32(F.col(pk).cast("string")) % 100 == 0)  # ~1% deterministic sample
                  .select(F.sha2(F.concat_ws("|", *[F.col(c).cast("string")
                                                    for c in df.columns]), 256).alias("h"))
                  .agg(F.sum(F.crc32("h")).alias("hsum")).first().hsum)
    checks["row_hash"] = sampled_hash(a) == sampled_hash(b)

    checks["PASS"] = all(checks.values())
    return checks


def cutover_ready(history: list[dict], n: int = 7) -> bool:
    """True only if the last n runs ALL passed."""
    return len(history) >= n and all(h["PASS"] for h in history[-n:])

Step-by-step explanation.

1. The harness reads the legacy output as raw Parquet and the candidate as the Iceberg table, so it compares the actual bytes each stack produced for the same partition — not two reads of the same source.
2. Tiers 1–3 are exact aggregate invariants computed in one pass each: count(*) catches missing or extra rows; sum(amount_cents) on integer cents is exact and catches value drift that leaves the count unchanged; count(distinct pk) catches duplicated or dropped primary keys that a raw count can hide.
3. Tier 4 is the deterministic sampled hash: it selects ~1% of rows by a stable function of the PK (crc32(pk) % 100 == 0, identical on both sides), hashes each sampled row's concatenated columns, and sums the hashes. If any column drifted on the sampled rows, the hash sums differ. This catches per-column corruption that aggregate checks miss, at 1% of the cost of hashing every row.
4. checks["PASS"] is the conjunction — every tier must pass. A run that passes counts and sums but fails the row hash is a fail, because it means a column drifted in a way the aggregates could not see.
5. cutover_ready enforces the N-consecutive-pass gate: it returns True only when the last n runs all passed. This rules out a single lucky pass and ensures the parity is stable across daily variation before any consumer is moved.

Output.

Run	count	sum	distinct	row_hash	PASS
1	ok	ok	ok	ok	true
2	ok	ok	ok	FAIL	false
3	ok	ok	ok	ok	true
4–9	ok	ok	ok	ok	true
gate (need 7)	—	—	—	—	not until run 9

Rule of thumb. Reconcile in tiers — count, exact integer sum, distinct keys, sampled row hash — and gate cutover on N consecutive all-tier passes, not one. The row hash is what catches per-column drift the aggregates hide, and the N-run gate is what rules out a lucky single pass.

Worked example — the wave cutover runbook with rollback

Detailed explanation. Cutover is executed table-by-table with a repeatable runbook: confirm the reconciliation gate, repoint the downstream consumer from the legacy table to the Iceberg table, keep the legacy output warm, and hold a rollback for the window. Walk through the runbook for one table.

• Precondition. Reconciliation gate green (N consecutive passes).
• Cutover. Repoint the consumer (e.g. a view or a config) from db.orders_legacy to the Iceberg db.orders.
• Rollback. Repoint back to legacy within the window if a problem surfaces.

Question. Write the cutover runbook as an idempotent, reversible operation for one table.

Input.

Step	Action	Reversible?
1	assert gate green	n/a
2	repoint consumer view to Iceberg	yes
3	keep legacy warm (dual-run continues)	yes
4	close window → stop legacy job	no (per table)

Code.

-- Consumers read a STABLE view name; cutover just repoints the view.
-- 1. Pre-cutover: the view points at the legacy Hadoop output
CREATE OR REPLACE VIEW mart.orders_current AS
SELECT * FROM hive.db.orders_legacy;

-- 2. CUTOVER: repoint the SAME view at the Iceberg table (atomic swap).
--    Downstream consumers change nothing — they still read mart.orders_current.
CREATE OR REPLACE VIEW mart.orders_current AS
SELECT * FROM glue.db.orders;        -- now the lakehouse table

-- 3. ROLLBACK (within window): repoint the view back to legacy.
--    Legacy job is still running (dual-run), so the data is fresh.
CREATE OR REPLACE VIEW mart.orders_current AS
SELECT * FROM hive.db.orders_legacy;

# 4. Runbook driver — gate-checked, reversible cutover for one table
def cutover_table(spark, table, history):
    if not cutover_ready(history, n=7):
        return f"HOLD {table}: reconciliation gate not green"
    spark.sql(f"""CREATE OR REPLACE VIEW mart.{table}_current
                  AS SELECT * FROM glue.db.{table}""")
    # legacy job keeps running for the rollback window; do NOT stop it yet
    return f"CUTOVER {table}: consumers now read Iceberg; legacy warm"

def rollback_table(spark, table):
    spark.sql(f"""CREATE OR REPLACE VIEW mart.{table}_current
                  AS SELECT * FROM hive.db.{table}_legacy""")
    return f"ROLLBACK {table}: consumers back on legacy (fresh via dual-run)"

Step-by-step explanation.

1. The key design move is the indirection view: consumers never reference db.orders or the Iceberg table directly — they read a stable mart.orders_current view. Cutover and rollback are just repointing that one view, so no consumer changes any code.
2. Pre-cutover, the view points at the legacy Hadoop output. Everything downstream is on Hadoop; the lakehouse table is being produced in parallel (dual-run) but nobody reads it yet.
3. Cutover is a single CREATE OR REPLACE VIEW that swaps the view's target to the Iceberg table. It is atomic at the catalog level and instantaneous; consumers on their next query transparently read the lakehouse table.
4. Crucially, the legacy job keeps running through the rollback window — the dual-run does not stop at cutover. That is what makes rollback safe: if a consumer reports a problem two days later, the legacy output is still fresh, and rollback_table repoints the view back with no data gap.
5. The gate is enforced in code: cutover_table refuses to repoint unless cutover_ready confirms N consecutive clean reconciles. This makes the runbook safe to hand to an operator — it cannot cut over a table that has not proven parity.

Output.

State	mart.orders_current points at	Legacy job	Reversible
Pre-cutover	legacy (Hadoop)	running	n/a
Cutover	Iceberg (lakehouse)	still running (warm)	yes
Rollback	legacy (Hadoop)	running	yes
Window closed	Iceberg	stopped	no (this table)

Rule of thumb. Put a stable indirection view between consumers and the physical table so cutover and rollback are one atomic CREATE OR REPLACE VIEW, and keep the legacy job running through the rollback window so a late-discovered problem is a repoint, not a rebuild. Gate the repoint on the reconciliation harness in code.

Worked example — the decommission checklist and realised cost saving

Detailed explanation. Decommission is the one irreversible step, executed once globally after every wave's rollback window closes. It has a strict order: drain YARN, prove no consumers remain on HDFS, archive a final manifest, then delete and power off. The moment the DataNodes go dark is when the projected cost saving becomes a realised one. Walk through the checklist.

• Drain. Stop new YARN scheduling; let running jobs finish.
• Prove empty. Assert no job references any HDFS path.
• Archive. Final checksum manifest of the HDFS warehouse to cold storage.
• Delete & power off. Delete HDFS; retire DataNodes; the cost saving is realised.

Question. Write the decommission checklist as a gated, ordered, mostly-irreversible runbook, and confirm the saving.

Input.

Step	Gate before it	Reversible?
Drain YARN	all waves past rollback window	yes
Prove no HDFS consumers	drain complete	yes
Archive manifest	zero consumers confirmed	yes
Delete HDFS + power off	manifest archived	NO

Code.

# 1. DRAIN — stop scheduling new work; let in-flight jobs finish
yarn rmadmin -refreshQueues                    # move all queues to drain state
yarn application -list -appStates RUNNING      # wait until this is empty

# 2. PROVE NO CONSUMERS — assert nothing still reads HDFS paths
#    (scan Airflow DAGs / job configs for hdfs:// references)
grep -R "hdfs://" /airflow/dags /jobs/ && echo "STILL REFERENCED — ABORT" || \
  echo "no hdfs:// references remain"

# 3. ARCHIVE — final manifest of the warehouse to cold storage (audit/restore)
hdfs dfs -ls -R /warehouse | \
  awk '{print $8, $5}' > /tmp/hdfs_manifest.txt
hdfs dfs -checksum /warehouse/* >> /tmp/hdfs_manifest.txt
aws s3 cp /tmp/hdfs_manifest.txt s3://lake-archive/decommission/manifest.txt

# 4. POINT OF NO RETURN — delete HDFS data, then power off DataNodes
#    Only after 1–3 are all green and every rollback window has closed.
hdfs dfs -rm -r -skipTrash /warehouse
# retire DataNodes via the exclude file + refreshNodes
echo "dn-hosts" > /etc/hadoop/dfs.exclude
hdfs dfsadmin -refreshNodes
# ... then power down the physical nodes -> COST SAVING REALISED HERE

# 5. Confirm the saving is now realised (both stacks no longer billed)
def realised_saving(before_monthly, after_monthly, dual_run_months, cutover_month):
    # During dual-run you paid BOTH; saving is realised only post-decommission
    return {
        "dual_run_cost_extra": before_monthly * dual_run_months,  # temporary overlap
        "steady_state_monthly_saving": before_monthly - after_monthly,
        "realised_from_month": cutover_month + 1,
    }

Step-by-step explanation.

1. Draining YARN (refreshQueues to a drain state, then waiting for RUNNING apps to empty) stops new work without killing in-flight jobs. This is reversible — you can un-drain — so it is safe to do first and observe.
2. Proving no consumers remain is the critical gate: a grep for hdfs:// across the Airflow DAGs and job configs must return nothing. A single lingering reference means a consumer would break the instant HDFS is deleted, so a hit aborts the decommission.
3. Archiving a final manifest — the file listing plus checksums, copied to cold object storage — is the audit trail and the last-resort restore point. It costs almost nothing and is the difference between "irreversible but documented" and "irreversible and blind."
4. Deleting HDFS (rm -r -skipTrash) and retiring the DataNodes via the exclude file is the point of no return. It is gated on steps 1–3 being green and every wave's rollback window having closed — because after this, there is no warm legacy to roll back to.
5. The saving becomes realised only here. During dual-run you paid for both stacks (a temporary, deliberate overlap cost); the steady-state monthly saving from Section 2's model only starts accruing once the DataNodes are actually off. This is why the rollback windows are time-boxed rather than open-ended.

Output.

Step	Gate	Reversible	Cost effect
Drain YARN	waves past window	yes	none
Prove no consumers	drain done	yes	none
Archive manifest	zero consumers	yes	trivial
Delete + power off	manifest archived	NO	saving realised

Rule of thumb. Decommission is a single ordered runbook gated on every rollback window closing: drain, prove-no-consumers (abort on any hdfs:// reference), archive a manifest, then delete and power off. The saving is only realised at power-off, so time-box the dual-run — every extra month of overlap is a month you pay for both stacks.

Senior interview question on cutover and decommission

A senior interviewer might ask: "Your data is copied, your tables are Iceberg, and your jobs are rewritten and dual-running. Now walk me through the cutover: how you reconcile the two stacks, how you cut consumers over without a big-bang, how you keep a rollback, and the exact gated sequence by which you finally decommission the Hadoop cluster and prove the cost saving. What is the one step you can never undo?"

Solution Using tiered reconciliation, view-indirection cutover, and a gated decommission

# 1. RECONCILE tiered (count -> integer sum -> distinct -> sampled hash),
#    gate cutover on N consecutive all-tier passes
def reconcile_and_gate(spark, legacy, candidate, pk, money, history, n=7):
    a, b = spark.read.parquet(legacy), spark.table(candidate)
    ra = a.agg(F.count("*").alias("c"), F.sum(money).alias("s"),
               F.countDistinct(pk).alias("d")).first()
    rb = b.agg(F.count("*").alias("c"), F.sum(money).alias("s"),
               F.countDistinct(pk).alias("d")).first()
    run = {"count": ra.c == rb.c, "sum": ra.s == rb.s, "distinct": ra.d == rb.d}
    run["PASS"] = all(run.values())
    history.append(run)
    return len(history) >= n and all(h["PASS"] for h in history[-n:])

-- 2. CUTOVER via view indirection (atomic, reversible)
CREATE OR REPLACE VIEW mart.orders_current AS SELECT * FROM glue.db.orders;
-- legacy job stays warm through the rollback window
-- ROLLBACK: CREATE OR REPLACE VIEW mart.orders_current AS
--           SELECT * FROM hive.db.orders_legacy;

# 3. DECOMMISSION — ordered, gated, last step (irreversible at the end)
yarn rmadmin -refreshQueues                                  # drain
grep -R "hdfs://" /airflow/dags /jobs/ && exit 1             # prove no consumers
hdfs dfs -ls -R /warehouse > /tmp/manifest.txt               # archive
aws s3 cp /tmp/manifest.txt s3://lake-archive/decommission/
hdfs dfs -rm -r -skipTrash /warehouse                        # POINT OF NO RETURN
echo dn-hosts > /etc/hadoop/dfs.exclude && hdfs dfsadmin -refreshNodes
# power down DataNodes -> cost saving realised

Step-by-step trace.

Phase	Mechanism	Reversible
Reconcile	count/sum/distinct/hash, N-run gate	n/a
Cutover	CREATE OR REPLACE VIEW → Iceberg	yes (repoint back)
Rollback window	legacy job stays warm	yes
Drain YARN	refreshQueues	yes
Prove no consumers	grep hdfs://	yes
Delete HDFS + power off	rm -r + refreshNodes	NO

After the reconciliation gate goes green for seven consecutive runs, each consumer is cut over by repointing a single indirection view at the Iceberg table, while the legacy job keeps running warm for the rollback window. Once every wave has cleared its window, the decommission runbook drains YARN, proves no job still references hdfs://, archives a final checksum manifest to cold storage, and only then deletes HDFS and powers off the DataNodes. The one step you can never undo is the final delete-and-power-off — which is exactly why it is gated on every prior check and why the cost saving is realised precisely there.

Output:

Metric	Value
Reconciliation gate	7 consecutive all-tier passes
Cutover mechanism	atomic view repoint
Rollback	repoint to warm legacy (minutes)
Irreversible step	delete HDFS + power off DataNodes
Saving realised	at DataNode power-off
Dual-run overlap	time-boxed (pays for both stacks)

Why this works — concept by concept:

• Tiered reconciliation + N-run gate — count, exact integer sum, distinct keys, and a sampled row hash catch progressively subtler drift, and requiring N consecutive clean runs rules out a lucky single pass. Cutover is gated on proof, not on optimism.
• View-indirection cutover — consumers read a stable view, so cutover and rollback are one atomic CREATE OR REPLACE VIEW. No consumer changes code, and rollback is a repoint rather than a rebuild.
• Warm legacy through the window — the dual-run does not stop at cutover; the legacy job stays warm so a late-discovered problem is a minutes-long repoint with no data gap. This is the safety net that makes each cutover reversible.
• Gated, ordered decommission — drain → prove-no-consumers → archive → delete is a strict order where only the last step is irreversible, and it is gated on every rollback window closing. A stray hdfs:// reference aborts the whole thing.
• Cost — dual-run temporarily pays for both stacks, so the projected saving is only realised at power-off; that is why the windows are time-boxed. The reconciliation is O(rows) aggregates plus a 1% hash sample — cheap insurance against shipping a silent correctness bug into a finance table.

ETL
Topic — etl
ETL problems on reconciliation and data validation

Practice →

Design
Topic — design
Design problems on zero-downtime cutover and rollback

Practice →

---

Cheat sheet — Hadoop → lakehouse migration recipes

• The four-layer map. A Hadoop migration is four independent migrations in dependency order: storage (HDFS → object store, tool: DistCp + S3A committer), table format (Hive → Iceberg, tool: snapshot/migrate/add_files), compute (MapReduce/HiveQL/Pig → Spark/Trino, tool: rewrite + golden diff), orchestration (Oozie → Airflow, tool: DAG rewrite). Sequence: copy storage → adopt table format → rewrite + dual-run jobs → reconcile + cut over → decommission last. Never rewrite jobs before data lands; never decommission before the last consumer moves.
• DistCp bulk + incremental template. Freeze an HDFS snapshot (hdfs dfs -createSnapshot /warehouse s0), bulk-copy (hadoop distcp -m 200 -bandwidth 15 -update -strategy dynamic hdfs://nn/warehouse s3a://lake/warehouse), then an exact catch-up with a second snapshot (distcp -update -diff s0 s1 ...), and a final micro-diff at cutover. Verify with per-prefix counts + byte totals — never cross-filesystem checksums (HDFS CRC32C ≠ S3 ETag).
• S3A committer config. Object stores have no atomic rename, so bind the magic committer: spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a=org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory, spark.hadoop.fs.s3a.committer.name=magic, plus spark.sql.sources.commitProtocolClass=...PathOutputCommitProtocol and the BindingParquetOutputCommitter. Commit becomes multipart completion (O(1) metadata), not an O(bytes) copy — or move to Iceberg and let the catalog pointer be the atomic commit.
• Iceberg adoption procedures. CALL cat.system.snapshot('db.hive_t','cat.db.ice_t') = independent shadow, source untouched (validate against it). CALL cat.system.migrate('db.t') = in-place cutover, keeps db.t_BACKUP_. CALL cat.system.add_files(table=>'db.ice_t', source_table=>'db.hive_t') = import files into an existing Iceberg table. All three adopt existing Parquet with zero data rewrite — adoption is O(files) metadata, not O(bytes).
• Iceberg maintenance Hive never needed. rewrite_data_files (compact the small files the object-store move created; target ~512 MB), expire_snapshots (reclaim orphaned files + trim metadata; retain last N), rewrite_manifests (keep pruning fast). Use hidden partitioning (days(ts), bucket(16,id)) so queries can't forget the partition predicate, and evolve the partition spec instead of rewriting to repartition.
• HiveQL → Spark SQL semantic-trap checklist. Pin these or parity silently breaks: session timezone (spark.sql.session.timeZone=UTC), strict casts (spark.sql.ansi.enabled=true), explicit CAST(... AS DECIMAL(p,s)) for money (never sum floats), NULL vs empty-string handling in text parsing, LATERAL VIEW explode → explode() (watch outer-explode on empty arrays), reserved words / identifier quoting. Classify jobs: HiveQL = light port, MapReduce/Pig = full DataFrame rewrite, Oozie = scheduling-only.
• Golden-output diff. Run legacy and rewritten jobs on the same frozen input; sort by keys; assert equal row count; full-outer-join and flag any measure where NOT (a.col <=> b.col) (null-safe). Zero differing rows = cutover gate PASS. A code read cannot catch decimal-rounding or NULL-handling drift; a byte diff can.
• Tiered reconciliation. count() → exact integer sum(money_cents) → count(distinct pk) → sampled deterministic row hash (crc32(pk)%100==0). Gate cutover on **N consecutive* all-tier passes (e.g. 7), never one. The row hash catches per-column drift the aggregates hide; the N-run gate rules out a lucky pass.
• View-indirection cutover. Consumers read a stable mart.t_current view; cutover = CREATE OR REPLACE VIEW mart.t_current AS SELECT * FROM iceberg.db.t (atomic); rollback = repoint the view back to warm legacy. Keep the legacy job running through the rollback window so rollback is a repoint, not a rebuild.
• Decommission checklist (irreversible — last step). Drain YARN (yarn rmadmin -refreshQueues, wait for RUNNING to empty) → prove no consumers (grep -R "hdfs://" dags/ jobs/ returns nothing, else ABORT) → archive final manifest + checksums to cold storage → delete HDFS (hdfs dfs -rm -r -skipTrash /warehouse) → retire DataNodes (dfs.exclude + refreshNodes) → power off. The cost saving is realised only at power-off.
• Cost model. Before = N DataNodes 24×7 (storage+compute fused, 3× replication). After = object storage per-GB-month (single erasure-coded copy) + autoscaled compute (minExecutors=0 off-peak). The saving is dominated by shedding idle compute and 3× replication — but it is only realised at decommission, so time-box the dual-run overlap where you pay for both stacks.
• What breaks first moving HDFS→object store. No atomic rename (jobs commit wrong), small-file explosion (every object is a GET — compact after copy), lost data locality (compute reads over the network — size executors for network, not disk), and cross-FS checksum mismatch (verify by counts/sizes/content, not CRC vs ETag).

Frequently asked questions

What does a "Hadoop to lakehouse migration" actually mean?

A Hadoop migration to a lakehouse is the coordinated retirement of four independent layers of a Hadoop estate onto a decoupled cloud stack: HDFS storage becomes an object store (S3 / ADLS / GCS), the Hive metastore's directory-listing tables become an open table format like Iceberg, MapReduce / Tez / Hive-on-YARN / Spark-on-YARN compute becomes Spark (batch) and Trino (interactive) on elastic compute, and Oozie orchestration becomes Airflow. The defining property of the lakehouse is that storage and compute are decoupled — you pay for object storage per GB-month independent of any cluster, and compute autoscales to zero between jobs — which is the opposite of HDFS's design, where storage and compute are fused on the same always-on DataNodes. The migration is risky not because any single layer is hard but because the four layers must be interleaved so no downstream consumer ever reads a half-migrated table, which is why every serious plan is wave-based, dual-run, and reconciled rather than a big-bang cutover.

Do I have to rewrite all my Parquet data to move to Iceberg?

No — and this is the single most important fact for a Hive to Iceberg migration at scale. Iceberg can adopt your existing Parquet files in place through three procedures: snapshot creates a new, independent Iceberg table that references the existing files while leaving the Hive source completely untouched (the safe validation baseline); migrate converts the Hive table to Iceberg in place and keeps a _BACKUP_ of the original; and add_files imports files from a Hive table or path into an existing Iceberg table. All three rewrite metadata, not data — Iceberg builds its manifest tree pointing at the Parquet files you already have, so adoption is an O(files) metadata operation that completes in minutes even for a huge table, not a second O(bytes) petabyte-scale rewrite. You get snapshots, ACID commits, hidden partitioning, schema evolution, and time-travel rollback for free. The only genuine data rewrite you should schedule is compaction (rewrite_data_files) after the move, to fix the small-file problem the HDFS-to-object-store copy inevitably creates.

HDFS to S3 — what breaks first?

The atomic rename breaks first. On HDFS, a job commits its output by renaming a _temporary staging directory into the final path, and that rename is a single atomic metadata operation. Object stores like S3 have a flat namespace with no rename — a "rename" is implemented as copy-every-object-then-delete, which is neither atomic nor cheap, so the classic Hadoop FileOutputCommitter can partially fail and leave a job that reports success with missing files. The fix is to bind an S3A committer (fs.s3a.committer.name=magic), which uses S3 multipart uploads and completes them at job-commit time — an atomic-per-file commit with no rename — or to move the table to Iceberg, whose commit swaps a single catalog pointer and never renames data files. After the rename issue, the next things to break are the small-file explosion (every object is a separate GET, so you must compact after copying), the loss of data locality (compute now reads over the network, so size executors for network throughput), and cross-filesystem checksum verification (HDFS CRC32C and S3 ETag are different algorithms — verify by counts, sizes, and content reconciliation, not checksums). Note that S3 read-after-write consistency is not a problem anymore — it has been strongly consistent since late 2020.

Is HiveQL compatible with Spark SQL?

Mostly, but "mostly" is where a Spark migration ships silent correctness bugs. HiveQL and Spark SQL share a dialect family, so the bulk of Hive queries port to Spark SQL with minor edits — which is why HiveQL jobs are the "light" class in a job-rewrite inventory. The parity risk lives in a short list of semantic differences: implicit type coercions that Hive and Spark handle differently (make casts explicit and enable spark.sql.ansi.enabled), NULL versus empty-string handling in text SerDes, DECIMAL precision and rounding (cast money explicitly and never sum as double), timezone defaults in functions like from_unixtime (pin spark.sql.session.timeZone), LATERAL VIEW explode becoming Spark's explode() (watch outer-explode on empty arrays), and reserved-word / identifier-quoting differences. The senior discipline is not to trust a code read but to prove each port with a golden-output diff: run the Hive job and the Spark job on the same frozen input and compare outputs with a null-safe, order-independent, exact comparison of every measure. MapReduce and Pig jobs are full rewrites into Spark DataFrame logic rather than HiveQL-style ports, and Oozie coordinators become Airflow DAGs where only the scheduling changes.

How do I validate a migration without a big-bang cutover?

You dual-run and reconcile. Both the legacy Hadoop pipeline and the new lakehouse pipeline produce every table in parallel on the same inputs, and a reconciliation harness compares their outputs in tiers: exact count(*) per table and partition, exact integer sum() of money columns (store money as integer cents so the sum is exact — never sum floats), count(distinct pk) to catch duplicated or dropped keys, and for the highest-value tables a sampled deterministic row hash that catches per-column drift the aggregates miss. Cutover is gated not on a single clean run but on N consecutive clean runs (commonly seven) to rule out a flaky pass. When the gate is green, you cut consumers over table-by-table using an indirection view — consumers read a stable mart.t_current view, and cutover is a single atomic CREATE OR REPLACE VIEW repointing it at the Iceberg table — while keeping the legacy job running warm for a rollback window. If a problem surfaces days later, rollback is a repoint of that one view back to the still-fresh legacy output, not a rebuild. This is why a lakehouse migration is never a big-bang: every table is independently proven, cut over, and reversible.

When can I actually decommission the Hadoop cluster?

Cluster decommission is the last step and the only irreversible one, so it is heavily gated. You may decommission only after every wave has been cut over and every wave's rollback window has closed — meaning no consumer can still need the warm legacy output. The runbook is strictly ordered: drain the YARN queues (yarn rmadmin -refreshQueues) and let in-flight jobs finish; prove no job still references any hdfs:// path (a grep across the Airflow DAGs and job configs must return nothing — a single hit aborts the decommission); archive a final checksum manifest of the HDFS warehouse to cold object storage for audit and last-resort restore; then delete the HDFS data (hdfs dfs -rm -r -skipTrash /warehouse), retire the DataNodes via the exclude file (dfs.exclude + hdfs dfsadmin -refreshNodes), and power down the physical nodes. The cost saving that justified the whole project is realised precisely at power-off, not before — during dual-run you were paying for both stacks — which is exactly why the rollback windows are time-boxed rather than open-ended: every extra month of overlap is a month you pay twice.

Practice on PipeCode

• Drill the ETL practice library → for the migration, incremental-load, reconciliation, and data-validation problems senior interviewers use to probe a lakehouse cutover.
• Rehearse on the data-processing practice library → for the Spark-rewrite, distributed-copy, and aggregation patterns that replace MapReduce and Hive-on-Tez.
• Sharpen the tuning axis with the optimization practice library → for the file-compaction, partitioning, and shuffle problems that make a post-move Iceberg table fast.
• Stack the modelling reps on the design practice library → for the phased-cutover, rollback, and platform-migration scenarios, and anchor the four-layer plan against PipeCode's broader 450+ data-engineering catalogue.

Lock in Hadoop-migration muscle memory

Docs explain the tools. PipeCode drills explain the decision — when to snapshot versus migrate a Hive table, why the object store's missing rename breaks your jobs, how a golden-output diff catches a decimal drift, and when you are finally allowed to power the DataNodes off. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice data-processing problems →