A data engineer's day: pipelines, migrations and the lakehouse

A price was wrong for a month

Pipelines, migrations and the lakehouse

Words you'll meet in this lesson

Architecture at a glance

A modern lakehouse on Hadoop or cloud storage. The green boxes are what we ran in this lesson; the rest is the usual company setup around it.

From our lab (Spark 4.0.4 + Iceberg 1.10.2 on Hadoop 3.5.0, YARN)

The job at three levels

Table formats, pipeline styles and migration routes

Pipeline and migration choices

Animated example: one day at a fintech

A lakehouse table on our Hadoop cluster

Spark 4.0.4 (Lesson 13.2) with the Iceberg runtime jar from Maven Central, on our YARN, writing to HDFS. The catalog is the simplest kind, a folder in HDFS (type=hadoop).

1. Create a partitioned Iceberg table and load it

   q("CREATE TABLE lake.canteen.orders (order_id BIGINT, student STRING, ts TIMESTAMP, item STRING, amount INT) "
     "USING iceberg PARTITIONED BY (days(ts))")
   src.writeTo("lake.canteen.orders").append()

   -- 1. loaded +------+--------+ |orders|revenue | +------+--------+ |500000|12852002| +------+--------+
   days(ts) is hidden partitioning: Iceberg derives the day from the timestamp. Queries filter on ts and Iceberg prunes days on its own.

2. Change rows with SQL

   DELETE FROM lake.canteen.orders WHERE item = 'coffee';
   UPDATE lake.canteen.orders SET amount = 18 WHERE item = 'chai';

   -- 2. after DELETE coffee + UPDATE chai price +------+--------+ |orders|revenue | +------+--------+ |459636|12493729| +------+--------+
   12,852,002 − 40,364 × 20 (coffee) + 149,669 × 3 (chai +₹3) = 12,493,729. On a plain folder of CSV or Parquet files this needs a full rewrite job.

3. Every change is a snapshot

   SELECT snapshot_id, operation, summary['added-records'], summary['deleted-records']
   FROM lake.canteen.orders.snapshots ORDER BY committed_at

   |6640816061555354185|append |500000|NULL | |4428609581335996210|overwrite|459636|500000 | |1149537224652247478|overwrite|459636|459636 |
   Copy-on-write: the DELETE rewrote every data file without the coffee rows (500,000 out, 459,636 in), and the UPDATE rewrote them again. Old files stay until the snapshots that use them are expired.

4. Time travel

   SELECT count(*), sum(amount) FROM lake.canteen.orders VERSION AS OF 6640816061555354185

   +------+--------+ |orders|revenue | +------+--------+ |500000|12852002| +------+--------+
   Finance's request from the story: last week's report can be reproduced exactly, while today's table is corrected.

5. Evolve the schema

   ALTER TABLE lake.canteen.orders ADD COLUMN payment STRING

   |order_id|item |amount|payment| |1 |dosa |45 |NULL | |2 |samosa|18 |NULL |
   A metadata-only change: no data file is rewritten. Iceberg tracks columns by ID, so rename and drop are safe too (compare Avro's rules in Lesson 10.1).

6. Look inside HDFS

   hdfs dfs -count /lakehouse/canteen/orders/data /lakehouse/canteen/orders/metadata

   31 90 7617041 /lakehouse/canteen/orders/data 1 14 75103 /lakehouse/canteen/orders/metadata v1.metadata.json v2.metadata.json v3.metadata.json snap-….avro (×3)
   It is still just HDFS: day folders of Parquet files plus JSON and Avro metadata. A commit is one atomic swap of the current metadata.json; that is where the transaction comes from.

SQL on the left, files in HDFS on the right

Six steps of our Iceberg run. Watch which files each statement adds and which snapshot the table points to.

Data engineering at work

Simplified; organisations differ in tools and policies.

Sources (checked 22 Oct 2026): Apache Attic: retired projects (Oozie Feb 2025, Sqoop Jun 2021) · Amazon EMR 7.x application versions (Iceberg 1.10.1, Delta 3.3.2, Hudi 1.0.2) · Cloudera Base 7.3.2 summary (Iceberg REST catalog sharing) · endoflife.date: Apache Spark

A table that keeps its history

A copy-on-write table in a few lines of Python: every change writes a new list of rows (new "files") and records a snapshot. The numbers are our lab's real totals, aggregated per item.

# per item: (orders, price) from our 500,000-order table
base = {"chai": (149669, 15), "coffee": (40364, 20), "dosa": (49927, 45), "idli": (44774, 30),
        "poha": (45336, 25), "samosa": (69884, 18), "thali": (30292, 80), "vada pav": (69754, 20)}
snapshots = []

def commit(op, rows):
    snapshots.append((len(snapshots) + 1, op, dict(rows)))     # copy-on-write: a whole new version

def stats(rows):
    return sum(n for n, p in rows.values()), sum(n * p for n, p in rows.values())

commit("append", base)
v2 = {k: v for k, v in snapshots[-1][2].items() if k != "coffee"}          # DELETE coffee
commit("overwrite", v2)
v3 = {k: (n, 18 if k == "chai" else p) for k, (n, p) in snapshots[-1][2].items()}   # UPDATE chai
commit("overwrite", v3)

for sid, op, rows in snapshots:
    print(sid, op.ljust(9), "orders %d, revenue %d" % stats(rows))
print("time travel to 1:", "orders %d, revenue %d" % stats(snapshots[0][2]))

• snapshots.append((len(snapshots) + 1, op, dict(rows))) # copy-on-write: a whole new versionLike Iceberg's copy-on-write: a change writes new data and a new snapshot; the old one is untouched.
• v3 = {k: (n, 18 if k == "chai" else p) for k, (n, p) in snapshots[-1][2].items()} # UPDATE chaiOur real UPDATE: chai from 15 to 18 rupees.
• print("time travel to 1:", "orders %d, revenue %d" % stats(snapshots[0][2]))VERSION AS OF the first snapshot gave 500,000 and 12,852,002 in the lab.

Run it on your PC: an Iceberg lakehouse on the Module 8 lab

Use the Module 8 lab with Spark 4.0.4 installed as in Lesson 13.2. Download the Iceberg runtime for Spark 4.0 (iceberg-spark-runtime-4.0_2.13-1.10.2.jar, 47 MB) from Maven Central (repo1.maven.org/maven2/org/apache/iceberg/) and copy it into the namenode container. The table lives in hdfs://namenode/lakehouse. Keep at least a few GB free on your disk: our first attempt failed because the disk was 97% full.

# lakehouse_iceberg.py: an Apache Iceberg table on HDFS (Spark 4.0 + Iceberg 1.10).
# Things a plain folder of Parquet files cannot do: row-level DELETE/UPDATE, snapshots, time travel,
# schema evolution - all as ordinary SQL, with the data still sitting in HDFS.
from pyspark.sql import SparkSession
spark = (SparkSession.builder.appName("lakehouse demo")
         .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
         .config("spark.sql.catalog.lake", "org.apache.iceberg.spark.SparkCatalog")
         .config("spark.sql.catalog.lake.type", "hadoop")                       # catalog = a folder in HDFS
         .config("spark.sql.catalog.lake.warehouse", "hdfs://namenode/lakehouse")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
q = lambda sql: spark.sql(sql)
def show(title, sql):
    print(f"-- {title}"); q(sql).show(truncate=False)

q("DROP TABLE IF EXISTS lake.canteen.orders PURGE")
q("CREATE TABLE lake.canteen.orders (order_id BIGINT, student STRING, ts TIMESTAMP, item STRING, amount INT) "
  "USING iceberg PARTITIONED BY (days(ts))")
src = (spark.read.csv("hdfs://namenode/user/hadoop/data/student_orders.csv", inferSchema=True)
       .toDF("order_id", "student", "ts", "item", "amount"))
src.writeTo("lake.canteen.orders").append()
show("1. loaded", "SELECT count(*) AS orders, sum(amount) AS revenue FROM lake.canteen.orders")
q("DELETE FROM lake.canteen.orders WHERE item = 'coffee'")                 # e.g. a supplier's orders were fake
q("UPDATE lake.canteen.orders SET amount = 18 WHERE item = 'chai'")       # price correction
show("2. after DELETE coffee + UPDATE chai price", "SELECT count(*) AS orders, sum(amount) AS revenue FROM lake.canteen.orders")
show("3. every change is a snapshot", "SELECT snapshot_id, operation, summary['added-records'] AS added, summary['deleted-records'] AS deleted FROM lake.canteen.orders.snapshots ORDER BY committed_at")
first = q("SELECT snapshot_id FROM lake.canteen.orders.snapshots ORDER BY committed_at LIMIT 1").first()[0]
show("4. time travel: the table as it was at the first snapshot",
     f"SELECT count(*) AS orders, sum(amount) AS revenue FROM lake.canteen.orders VERSION AS OF {first}")
q("ALTER TABLE lake.canteen.orders ADD COLUMN payment STRING")
show("5. schema evolution: a new column, old rows read as NULL", "SELECT order_id, item, amount, payment FROM lake.canteen.orders ORDER BY order_id LIMIT 3")
show("6. partitions Iceberg manages for us (first 3 days)", "SELECT partition, record_count, file_count FROM lake.canteen.orders.partitions ORDER BY partition LIMIT 3")
spark.stop()

# Inside lab8-namenode-1 (SPARK_HOME = where you unpacked Spark 4.0.4):
#   export HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop
#   $SPARK_HOME/bin/spark-submit --master yarn --num-executors 3 --executor-memory 768m \
#     --executor-cores 1 --driver-memory 768m \
#     --jars iceberg-spark-runtime-4.0_2.13-1.10.2.jar lakehouse_iceberg.py
#   hdfs dfs -count /lakehouse/canteen/orders/data /lakehouse/canteen/orders/metadata

A backfill that cannot double count

Three days are loaded, day 2 is reloaded after a fix. Watch the partition dictionary: the reload replaces day 2 instead of adding to it.

Worked examples: real problems, solved and traced in memory

Four everyday data-engineering calculations from different industries.

45 days · 12 min/day · 5 parallel

9 waves, 108 minutes, done at 20:48

900 jobs · 40/week · 2-week dual-run · 6-week copy

jobs rewritten by week 23, last cut-over week 25

counts and totals from both systems

counts match: False | missing: 12
money diff: 0.012%

8 TB · 5% rewritten/day · 30 vs 7 days

30 days: 12.0 TB of old files (150% of the table)
 7 days: 2.8 TB of old files (35% of the table)

Check yourself

Quick revision

Interview and viva questions

Practice problems

Data-engineering exercises using our lakehouse run and everyday scenarios.

12,852,002 · 40,364 coffee at ₹20 · 149,669 chai +₹3

12493729

print(12852002 - 40364 * 20 + 149669 * 3)

500,000 + 459,636 + 459,636

1419272

print(500000 + 459636 + 459636)

31 dirs incl. data/ itself

30 days (September 2026)

print(31 - 1, "days (September 2026)")

a list of amounts

2 bad rows (20%) FAIL

amounts = [15, 20, None, 45, -18, 30, 25, 18, 20, 80]
bad = sum(1 for a in amounts if a is None or a < 0)
share = bad / len(amounts)
print(f"{bad} bad rows ({share:.0%})", "FAIL" if share > 0.01 else "OK")

two schemas

added: ['bank_ref'] removed: []

old = ["txn_id", "payer", "payee", "amount", "ts"]
new = ["txn_id", "payer", "payee", "amount", "ts", "bank_ref"]
print("added:", [c for c in new if c not in old], "removed:", [c for c in old if c not in new])

1 commit/min · 12 files/commit

17280 files/day -> 288 after hourly compaction

per_day = 60 * 24 * 12
print(per_day, "files/day ->", 24 * 12, "after hourly compaction")

12 TB · ₹1,800/TB/month

Rs 198,720 per year

saved_tb = 12 * (1 - 7 / 30)
print(f"Rs {saved_tb * 1800 * 12:,.0f} per year")

50 updates × 20 files

copy-on-write: 1000 files/day | merge-on-read: 70 files/day

cow = 50 * 20
mor = 50 + 20                 # 50 small delete files + nightly compaction rewriting about 20 files
print("copy-on-write:", cow, "files/day | merge-on-read:", mor, "files/day")

Real-world scenario questions (asked in company interviews)

Data-engineering interviews mix pipeline design, debugging stories and lakehouse questions.