Build an Iceberg Lakehouse with PySpark
Build a local data lakehouse with Apache Iceberg, PySpark, and MinIO.
Introduction
30 Second Summary
Every time you add a column to a database table or fix a batch of bad records, traditional data systems force you to rewrite everything. At production scale, that means reprocessing terabytes of data for a single change.
In this project, you will build a local data lakehouse powered by Apache Iceberg, PySpark, and MinIO. You will load e-commerce data into Iceberg tables and explore the features hiring managers care about most: schema evolution, time travel, and upserts.
What You'll Build
A local lakehouse environment running entirely on your machine, storing e-commerce order data in S3-compatible object storage and managing it through Iceberg's table format.
By the end of this project, you'll have:
- A Docker-based lakehouse stack with Spark, an Iceberg REST catalog, and MinIO object storage.
- PySpark code that creates, loads, and evolves Iceberg tables with schema and partition changes.
- Hands-on experience with time travel queries, upsert operations using MERGE INTO, and table maintenance that interviewers ask about.
- Secret Mission: Build a Bronze/Silver/Gold medallion architecture on top of your lakehouse.
Are there any prerequisites?
This project assumes familiarity with Python, SQL, and basic Spark concepts. You will also need Docker Desktop installed on your machine.
The project uses Docker Compose to run all services locally, so no other installations are needed.
Before We Start
Before We Start...
Set Up the Lakehouse Environment
A lakehouse needs several components working together: a query engine to process data, a catalog to track tables, and object storage to hold the actual files. Setting each one up individually and wiring them together would take hours of configuration.
In this step, you will spin up a complete lakehouse stack using Docker Compose. One YAML file defines four pre-configured services, and a single terminal command brings them all online.
In this step, get ready to:
- Create a project folder and a docker-compose.yml file that defines your lakehouse services.
- Start all four Docker containers with one command.
- Verify that Jupyter, MinIO, and the Iceberg catalog are running.
Create the project directory and Docker Compose file
Docker Compose lets you define multiple services in a single YAML file. Your lakehouse needs four services: a Spark query engine with Apache Iceberg pre-installed, a REST catalog, MinIO object storage, and a MinIO client that creates the initial storage bucket.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir iceberg-lakehouse
cd iceberg-lakehouse- Create an empty docker-compose.yml file by running this command:
macOS
touch docker-compose.ymlWindows
New-Item docker-compose.yml- Confirm the file was created by running:
lsYou should see docker-compose.yml in the output.
- Open docker-compose.yml in your preferred text editor, then paste the following configuration into it:
services:
spark-iceberg:
image: tabulario/spark-iceberg
container_name: spark-iceberg
build: spark/
networks:
iceberg_net:
depends_on:
- rest
- minio
volumes:
- ./warehouse:/home/iceberg/warehouse
- ./notebooks:/home/iceberg/notebooks/notebooks
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
ports:
- 8888:8888
- 8080:8080
- 10000:10000
- 10001:10001
rest:
image: apache/iceberg-rest-fixture
container_name: iceberg-rest
networks:
iceberg_net:
ports:
- 8181:8181
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
- CATALOG_WAREHOUSE=s3://warehouse/
- CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
- CATALOG_S3_ENDPOINT=http://minio:9000
minio:
image: minio/minio
container_name: minio
environment:
- MINIO_ROOT_USER=admin
- MINIO_ROOT_PASSWORD=password
- MINIO_DOMAIN=minio
networks:
iceberg_net:
aliases:
- warehouse.minio
ports:
- 9001:9001
- 9000:9000
command: ["server", "/data", "--console-address", ":9001"]
mc:
depends_on:
- minio
image: minio/mc
container_name: mc
networks:
iceberg_net:
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
entrypoint: |
/bin/sh -c "
until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
/usr/bin/mc rm -r --force minio/warehouse;
/usr/bin/mc mb minio/warehouse;
/usr/bin/mc policy set public minio/warehouse;
tail -f /dev/null
"
networks:
iceberg_net:What do these four services do?
- spark-iceberg runs PySpark with Iceberg pre-configured. It also serves Jupyter Notebook on port 8888 so you can write code in your browser.
- rest is the Iceberg REST catalog. It keeps track of which tables exist, where their data lives, and what schema each table has.
- minio provides S3-compatible object storage. Iceberg stores its Parquet data files and metadata here, exactly the same way a production lakehouse stores data on AWS S3.
- mc is the MinIO client. It runs once at startup to create the warehouse bucket that Iceberg needs for storage.
- Save docker-compose.yml.
Seeing a YAML syntax error later?
YAML is whitespace-sensitive. Make sure you pasted the entire file without extra or missing spaces. Each indentation level uses exactly two spaces, not tabs.
Double-check that the file is named exactly docker-compose.yml with no extra extensions like .txt.
Help me debug a YAML error in my docker-compose.yml file.
✔️ Awesome, I've got everything!
Double-check that you have saved docker-compose.yml before moving on.
ⓧ I'd like to double check the full code
Here is the complete docker-compose.yml file. Make sure yours matches exactly.
services:
spark-iceberg:
image: tabulario/spark-iceberg
container_name: spark-iceberg
build: spark/
networks:
iceberg_net:
depends_on:
- rest
- minio
volumes:
- ./warehouse:/home/iceberg/warehouse
- ./notebooks:/home/iceberg/notebooks/notebooks
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
ports:
- 8888:8888
- 8080:8080
- 10000:10000
- 10001:10001
rest:
image: apache/iceberg-rest-fixture
container_name: iceberg-rest
networks:
iceberg_net:
ports:
- 8181:8181
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
- CATALOG_WAREHOUSE=s3://warehouse/
- CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
- CATALOG_S3_ENDPOINT=http://minio:9000
minio:
image: minio/minio
container_name: minio
environment:
- MINIO_ROOT_USER=admin
- MINIO_ROOT_PASSWORD=password
- MINIO_DOMAIN=minio
networks:
iceberg_net:
aliases:
- warehouse.minio
ports:
- 9001:9001
- 9000:9000
command: ["server", "/data", "--console-address", ":9001"]
mc:
depends_on:
- minio
image: minio/mc
container_name: mc
networks:
iceberg_net:
environment:
- AWS_ACCESS_KEY_ID=admin
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
entrypoint: |
/bin/sh -c "
until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
/usr/bin/mc rm -r --force minio/warehouse;
/usr/bin/mc mb minio/warehouse;
/usr/bin/mc policy set public minio/warehouse;
tail -f /dev/null
"
networks:
iceberg_net:Start the Docker containers
With the Docker Compose file saved, one command pulls all four container images and starts them. The first run downloads around 2-3 GB of images, so it may take a few minutes depending on your internet speed.
- Make sure Docker Desktop is running on your machine.
- Start all four services by running this command in the same terminal (inside your iceberg-lakehouse folder):
docker-compose up -dWhat does this command do?
The -d flag runs the containers in detached mode, meaning they run in the background and give you your terminal back. Without it, your terminal would be locked showing container logs.
- Wait for the command to finish. You will see each service being created and started.
- Verify all containers are running by running:
docker psYou should see four containers listed: spark-iceberg, iceberg-rest, minio, and mc. All should have a status of Up.
Containers not starting?
Make sure Docker Desktop is running before you run docker-compose up -d. If Docker is not running, you will see a "Cannot connect to the Docker daemon" error.
If a port conflict prevents a container from starting, check that nothing else is using ports 8888, 9000, 9001, or 8181. Stop the conflicting application and run docker-compose up -d again.
Help me debug why my Docker containers are not starting.
Verify the services and create your notebook
Three of the four services expose web interfaces you can open in your browser. The fourth service (mc) already finished its job by creating the storage bucket at startup.
- Open your browser and navigate to http://localhost:8888.
You should see the Jupyter Notebook interface with a file browser. This is where you will write all your PySpark code.
- Open a new browser tab and navigate to http://localhost:9001.
- Log in with username admin and password password.
- Click Object Browser in the left sidebar.
You should see a bucket called warehouse listed. This is the S3-compatible storage where Iceberg will write all your table data and metadata.
Why MinIO instead of real S3?
MinIO provides an S3-compatible API that runs locally. Iceberg stores data as Parquet files and metadata as JSON in MinIO buckets, exactly mirroring how a production lakehouse works on AWS S3.
This means the PySpark code you write here would work against real S3 with only a configuration change. No code rewrite needed.
Now create the notebook where you will write all your PySpark code for the rest of this project.
- Switch back to the Jupyter tab at http://localhost:8888.
- Click New in the top-right corner of the file browser.
- Select Python 3 (or Notebook) from the dropdown.
- Click the notebook title at the top of the page (it defaults to Untitled) and rename it to lakehouse.
- Press Enter to confirm the rename.
You should now see lakehouse as the notebook name in the top bar. The file is saved as lakehouse.ipynb.
Can't reach localhost:8888 or localhost:9001?
Go back to your terminal and run docker ps to check that all four containers show a status of Up. If a container exited, run docker-compose down followed by docker-compose up -d to restart everything.
The spark-iceberg container can take 30-60 seconds to fully start on first launch. Wait a moment and try refreshing the page.
Help me troubleshoot why I can't connect to Jupyter or MinIO.
Your lakehouse environment is up and running. Next up, you will create your first Iceberg table and load it with data.
Create Iceberg Tables and Load Data
Your lakehouse stack is humming along with all four containers running. Now it is time to put it to work by creating your first Iceberg table and filling it with data.
In this step, you will create a namespace and partitioned table inside your Jupyter notebook using PySpark. Then you will generate 500 synthetic e-commerce orders, load them into the table, and verify everything landed correctly in MinIO.
In this step, get ready to:
- Create an Iceberg namespace and a partitioned orders table.
- Generate and load 500 synthetic e-commerce orders.
- Query the data and inspect the table's metadata in MinIO.
Create a namespace and table
Iceberg organizes tables under namespaces, similar to schemas in a traditional database. You will create a namespace called demo.ecommerce and then define a table called orders inside it.
Before writing any SQL, you need to import the PySpark functions you will use throughout this project.
- Open your lakehouse.ipynb notebook in Jupyter at http://localhost:8888.
- Paste the following imports into the first cell:
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetimeWhat do these imports do?
The pyspark.sql.functions imports give you column manipulation tools like col, random data generators like rand, and string formatters like concat. You will use these to build synthetic order data.
The pyspark.sql.types imports define column schemas. You will need them later when creating DataFrames with explicit types.
- Run the cell by pressing Shift+Enter. You should see no output and no errors.
Seeing a ModuleNotFoundError?
This means PySpark is not available in your notebook kernel. Make sure you are running the notebook inside the Jupyter interface at http://localhost:8888, not a local Jupyter installation.
Confirm the spark-iceberg container is running with docker ps in your terminal.
Help me fix my PySpark import error in the Iceberg Docker container.
Now you will create the namespace and table. The demo catalog is pre-configured in the Docker image as a REST catalog. All table references follow the pattern demo.namespace.table.
- Create a new cell below and paste this code to set up your namespace and table:
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")What does this code do?
- CREATE NAMESPACE IF NOT EXISTS demo.ecommerce creates a namespace (like a schema) to organize your tables under.
- CREATE TABLE defines the orders table with six columns: order_id, customer_id, product_name, quantity, price, and order_date.
- USING iceberg tells Spark to create an Iceberg table (not a default Hive table).
- PARTITIONED BY (days(order_date)) partitions data by day. This means Spark groups rows with the same date into the same data files, making date-range queries much faster.
Where does the spark variable come from?
The spark variable is a pre-configured SparkSession that the Docker image makes available automatically in every notebook. It already has the Iceberg catalog, SQL extensions, and MinIO connection set up for you.
- Run the cell by pressing Shift+Enter.
- You should see Table demo.ecommerce.orders created successfully. printed below the cell.
Seeing a NameError for spark?
Make sure you are using the notebook inside the Docker-hosted Jupyter at http://localhost:8888. The spark variable is only available in that environment. If you opened a different Jupyter instance, it will not have PySpark configured.
Help me fix my NameError for spark in my Iceberg notebook.
Generate and load synthetic data
An empty table is not very useful. You need data to work with for the rest of this project.
You will use PySpark's DataFrame API to generate 500 synthetic e-commerce orders with randomized products, prices, and dates spread across the last 30 days.
- Create a new cell below and paste this code to generate and load the synthetic data:
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")What does this code do?
- spark.range(0, 500) creates 500 rows with IDs from 0 to 499. Each row becomes one order.
- The .select() block transforms each row into a realistic order with a formatted order_id (ORD-0000 to ORD-0499), a random customer_id (CUST-001 to CUST-050), and a random product from five options.
- .writeTo("demo.ecommerce.orders").append() uses Iceberg's v2 write API to append the data to your table. This creates a new snapshot, which is how Iceberg tracks every change.
- Run the cell by pressing Shift+Enter. This may take 10-20 seconds as Spark writes the data files.
- You should see Loaded 500 orders into demo.ecommerce.orders. printed below the cell.
Seeing a AnalysisException or table not found error?
Make sure you ran Cell 2 (Create Namespace and Table) successfully before running this cell. The table must exist before you can append data to it.
Also confirm you ran Cell 1 (Imports) first. The functions like concat, lit, and rand need to be imported before you can use them.
Help me fix my data loading error in my Iceberg notebook.
Query and inspect the table
Your data is loaded. Now you will query it to confirm everything looks right and inspect the Iceberg metadata that was created behind the scenes.
- Create a new cell below and paste this code to query and inspect your table:
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)What do these queries do?
- The first query previews the first 10 orders sorted by order_id so you can see the data shape.
- The second query counts total rows to confirm all 500 orders were loaded.
- The third query reads the .snapshots metadata table. Every time you change an Iceberg table, it creates a new snapshot. You should see exactly one snapshot from the append you just ran.
- Run the cell by pressing Shift+Enter.
- You should see three tables printed: a preview of 10 orders, a count showing 500 total orders, and a single snapshot with its snapshot_id, committed_at timestamp, and append as the operation.
Seeing fewer than 500 rows?
If you ran Cell 3 more than once, you may see more than 500 rows because .append() adds data without checking for duplicates. To reset, drop the table with spark.sql("DROP TABLE demo.ecommerce.orders"), then re-run Cell 2 and Cell 3.
Help me fix my order count in my Iceberg table.
Now check what Iceberg actually wrote to storage. Under the hood, Iceberg stores your data as Parquet files with JSON and Avro metadata files alongside them in MinIO.
- Open the MinIO console at http://localhost:9001 in your browser.
- Log in with username admin and password password.
- Click on the warehouse bucket in the left sidebar.
- Browse into the ecommerce folder, then into orders.
- You should see a data folder (containing Parquet files) and a metadata folder (containing JSON and Avro files that track your table's schema, partitions, and snapshots).
Why does this matter?
This is exactly how production Iceberg tables look on AWS S3, Google Cloud Storage, or Azure Blob Storage. The only difference is the URL. Everything you learn here transfers directly to cloud deployments.
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook.
ⓧ I'd like to double check the full code
Here is the complete notebook code across all four cells. Compare it against your lakehouse.ipynb to make sure everything matches.
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetime
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)Your Iceberg table is loaded with 500 orders and verified end-to-end. Next up, you will put Iceberg's superpowers to the test by evolving the table's schema and partition strategy without rewriting a single data file.
Evolve Your Schema and Partitions
Your lakehouse has 500 orders and a working Iceberg table. Now imagine your e-commerce team needs a new column to track discounts, and the analytics team wants to switch from daily to monthly partitions for faster queries.
In a traditional data warehouse, both of these changes would require rewriting every row of data. Iceberg handles them as lightweight metadata updates, with zero data rewriting.
In this step, get ready to:
- Add a new column to the orders table without rewriting any existing data.
- Change the table's partitioning strategy from daily to monthly.
Add a column with schema evolution
Your orders table currently has six columns. You need to add a discount column so new orders can track promotional pricing.
What is schema evolution?
Schema evolution is the ability to change a table's column structure (add, drop, rename, reorder) without rewriting existing data files.
Iceberg makes this possible because it tracks columns by unique field IDs, not by name or position. An ALTER TABLE ADD COLUMNS command only updates the metadata. Existing Parquet data files stay untouched.
- In your lakehouse.ipynb notebook, create a new cell and add this code to add the discount column and insert 5 new orders that use it:
# =============================================================
# Cell 5: Schema Evolution - Add Column (Step 3)
# =============================================================
# Add a discount column without rewriting existing data
spark.sql("ALTER TABLE demo.ecommerce.orders ADD COLUMNS (discount DOUBLE)")
print("Added 'discount' column to orders table.")
# Insert 5 new orders that include discount values
spark.sql("""
INSERT INTO demo.ecommerce.orders VALUES
('ORD-0500', 'CUST-010', 'Laptop', 2, 999.99, current_timestamp(), 0.10),
('ORD-0501', 'CUST-020', 'Monitor', 1, 349.99, current_timestamp(), 0.15),
('ORD-0502', 'CUST-030', 'Keyboard', 3, 79.99, current_timestamp(), 0.05),
('ORD-0503', 'CUST-040', 'Headphones', 1, 199.99, current_timestamp(), 0.20),
('ORD-0504', 'CUST-050', 'Mouse', 4, 29.99, current_timestamp(), 0.00)
""")
print("Inserted 5 new orders with discount values.")
# Verify: old rows have NULL discount, new rows have values
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0000', 'ORD-0001', 'ORD-0500', 'ORD-0501')
ORDER BY order_id
""").show()What does this code do?
- ALTER TABLE ... ADD COLUMNS adds a discount column of type DOUBLE. This is a metadata-only operation. No existing data files are rewritten.
- INSERT INTO adds 5 new orders (ORD-0500 through ORD-0504) that include actual discount values.
- The final SELECT compares old orders against new ones. Old rows will show NULL for discount because they were written before the column existed.
- Run the cell by pressing Shift+Enter.
You should see a table with four rows. ORD-0000 and ORD-0001 show NULL in the discount column, while ORD-0500 and ORD-0501 show actual discount values like 0.1 and 0.15.
Seeing a ColumnAlreadyExistsException?
- This means you already ran this cell once. The discount column already exists. You can safely skip the ALTER TABLE line by commenting it out, then re-run.
- If you also see duplicate orders, the INSERT INTO ran more than once. You can verify by checking the row count with SELECT count(*) FROM demo.ecommerce.orders. It should be 505.
Help me troubleshoot my schema evolution cell.
Change the partitioning strategy
Your orders table is currently partitioned by day. As the table grows, daily partitions become too granular and create thousands of tiny files.
What is partition evolution?
Partition evolution lets you change how a table is partitioned without rewriting historical data. Old data stays in its original daily partitions. New data uses the new monthly partitions.
In Hive-style partitioning, changing the partition scheme requires a full data rewrite. Iceberg tracks partition specs as versioned metadata, so the switch is instant.
- Create a new cell in your notebook and add this code to switch from daily to monthly partitioning:
# =============================================================
# Cell 6: Partition Evolution (Step 3)
# =============================================================
# Switch from daily to monthly partitions
spark.sql("""
ALTER TABLE demo.ecommerce.orders
REPLACE PARTITION FIELD days(order_date) WITH months(order_date)
""")
print("Partition strategy changed from days(order_date) to months(order_date).")
# Inspect the current partition layout
spark.sql("SELECT * FROM demo.ecommerce.orders.partitions").show(truncate=False)What does this code do?
- REPLACE PARTITION FIELD swaps the partition transform from days(order_date) to months(order_date). Historical data keeps its daily partition layout.
- The .partitions metadata table shows every partition in the table. You will see a mix of daily and monthly partitions coexisting.
- Run the cell by pressing Shift+Enter.
You should see the confirmation message followed by a table of partition data. The output shows partition values alongside file and record counts for each partition.
Seeing an AnalysisException?
- Double-check that the old partition field is spelled exactly as days(order_date). If you already ran this cell once, the field was already replaced and you will see an error because days(order_date) no longer exists.
- If you have already switched to monthly partitions, you can skip this cell. The change persists across cell re-runs.
Help me troubleshoot my partition evolution cell.
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook.
ⓧ I'd like to double check the full code
Here is the full notebook code through the end of this step. Compare it cell by cell against your lakehouse.ipynb.
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetime
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)
# =============================================================
# Cell 5: Schema Evolution - Add Column (Step 3)
# =============================================================
spark.sql("ALTER TABLE demo.ecommerce.orders ADD COLUMNS (discount DOUBLE)")
print("Added 'discount' column to orders table.")
spark.sql("""
INSERT INTO demo.ecommerce.orders VALUES
('ORD-0500', 'CUST-010', 'Laptop', 2, 999.99, current_timestamp(), 0.10),
('ORD-0501', 'CUST-020', 'Monitor', 1, 349.99, current_timestamp(), 0.15),
('ORD-0502', 'CUST-030', 'Keyboard', 3, 79.99, current_timestamp(), 0.05),
('ORD-0503', 'CUST-040', 'Headphones', 1, 199.99, current_timestamp(), 0.20),
('ORD-0504', 'CUST-050', 'Mouse', 4, 29.99, current_timestamp(), 0.00)
""")
print("Inserted 5 new orders with discount values.")
# Verify: old rows have NULL discount, new rows have values
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0000', 'ORD-0001', 'ORD-0500', 'ORD-0501')
ORDER BY order_id
""").show()
# =============================================================
# Cell 6: Partition Evolution (Step 3)
# =============================================================
spark.sql("""
ALTER TABLE demo.ecommerce.orders
REPLACE PARTITION FIELD days(order_date) WITH months(order_date)
""")
print("Partition strategy changed from days(order_date) to months(order_date).")
spark.sql("SELECT * FROM demo.ecommerce.orders.partitions").show(truncate=False)Your table now has a new column and a new partitioning strategy, all without a single data rewrite. Next up, you will use Iceberg's time travel to query the table as it looked before any of these changes.
Query Historical Snapshots with Time Travel
Your table has been through a lot already. You added a column, inserted new rows, and changed the partitioning strategy.
Every one of those changes created a snapshot, a point-in-time version of the table that Apache Iceberg keeps around. In this step, you will use time travel to query the table as it existed before any of your schema or partition changes. This is a key feature for auditing, debugging, and reproducing past results in production.
In this step, get ready to:
- List every snapshot Iceberg has recorded for your orders table.
- Travel back in time to query the original 500-row table before schema evolution.
- Inspect the full change history log for auditing and compliance.
List all snapshots
Iceberg automatically records a snapshot every time data or metadata changes. Each snapshot captures the table's state at that exact moment, including which data files belong to it and what schema it used.
- In your lakehouse.ipynb notebook, create a new cell and query the snapshots metadata table by pasting this code:
# =============================================================
# Cell 7: List Snapshots (Step 4)
# =============================================================
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at
""").show(truncate=False)What does this code do?
Every Iceberg table has a hidden .snapshots metadata table that tracks all versions.
- snapshot_id is a unique long integer that identifies each version of the table.
- committed_at records the exact timestamp when that change was committed.
- operation tells you what happened (e.g. append for new data, overwrite for replacements).
- summary contains a map of details like how many records were added or removed.
- Run the cell by pressing Shift+Enter.
- You should see at least 3 snapshots listed in chronological order: the initial 500-row append, the 5-row insert with discounts, and the partition evolution.
Seeing an empty table or error?
- Make sure the table name is exactly demo.ecommerce.orders.snapshots with the .snapshots suffix. Missing the suffix queries the data table instead.
- Confirm you ran all previous cells (Cells 1 through 6) in order. Snapshots only exist if the prior steps completed successfully.
Still stuck? Ask for help.
Travel back in time
You now know every snapshot that exists. The first snapshot captures the table right after the initial 500-row load, before you added the discount column or changed the partitioning. You can query the table exactly as it was at that moment using VERSION AS OF.
- Create a new cell in your notebook and paste this code to extract the first snapshot ID and compare past vs present:
# =============================================================
# Cell 8: Time Travel Query (Step 4)
# =============================================================
# Grab the very first snapshot ID from the metadata table
first_snapshot_id = spark.sql("""
SELECT snapshot_id
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at ASC
LIMIT 1
""").collect()[0][0]
print(f"First snapshot ID: {first_snapshot_id}")
# Query the table as of the first snapshot
spark.sql(f"""
SELECT count(*) AS historical_row_count
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
""").show()
# Compare with current count
spark.sql("SELECT count(*) AS current_row_count FROM demo.ecommerce.orders").show()
# Check that the old snapshot has no discount column
spark.sql(f"""
SELECT *
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
LIMIT 5
""").show()What does this code do?
- The first query grabs the oldest snapshot_id from the snapshots metadata table and stores it in a Python variable called first_snapshot_id. The .collect()[0][0] pattern pulls a single value out of a PySpark DataFrame into plain Python.
- VERSION AS OF {first_snapshot_id} tells Iceberg to read the table as it existed when that snapshot was recorded. The snapshot ID is a long integer used directly in SQL (no quotes needed).
- The three queries compare the historical snapshot (500 rows, no discount column) with the current table (505 rows, has discount column). The data was never rewritten. Iceberg just reads the old files with the old schema.
- Run the cell by pressing Shift+Enter.
- You should see historical_row_count of 500 and current_row_count of 505.
- The final SELECT * output should show rows with only 6 columns (no discount column), confirming you are looking at the pre-evolution version of the table.
Why is time travel useful in production?
Imagine a bad data load corrupts thousands of rows in your production table at 2 AM. With time travel, you can immediately query the table as it was before the corruption to verify what changed. You can also reproduce the exact dataset a report used last Tuesday for auditing or compliance.
This is possible because Iceberg never modifies existing data files. Each snapshot points to a specific set of files, so reading an old snapshot simply reads the old files.
Seeing an AnalysisException or wrong row count?
- If you see an error about an invalid snapshot ID, make sure Cell 7 ran successfully first. The first_snapshot_id variable depends on the snapshots metadata table being populated.
- If historical_row_count shows 505 instead of 500, your first snapshot might not be the earliest one. Check the printed snapshot ID and compare it to the ordered list from Cell 7.
Need a hand debugging?
Inspect the change history
Snapshots show you individual versions. The .history metadata table goes further by showing the chronological chain of changes, including which snapshot is the current ancestor and how each snapshot relates to the one before it.
- Create a new cell and query the history metadata table by pasting this code:
# =============================================================
# Cell 9: View Change History (Step 4)
# =============================================================
# Show the full chronological log of table changes
spark.sql("SELECT * FROM demo.ecommerce.orders.history").show(truncate=False)What does the history table show?
The .history metadata table includes columns for made_current_at, snapshot_id, parent_id, and is_current_ancestor.
The parent_id column links each snapshot to the one that came before it, forming a chain. The is_current_ancestor column tells you whether that snapshot is part of the lineage that leads to the current table state. In production, this is what you use for auditing and compliance.
- Run the cell by pressing Shift+Enter.
- You should see a table with one row per snapshot, ordered chronologically. Every row should show true in the is_current_ancestor column since all snapshots are part of the current lineage.
Seeing an error with the history table?
- Double check the metadata table suffix is .history (not .histories or .changelog).
- Make sure the full table path is demo.ecommerce.orders.history with the demo catalog prefix included.
Ask for help if you are stuck.
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook before moving on.
ⓧ I'd like to double check the full code
Here is the full lakehouse.ipynb notebook up to this point. Compare each cell against yours.
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetime
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)
# =============================================================
# Cell 5: Schema Evolution - Add Column (Step 3)
# =============================================================
spark.sql("ALTER TABLE demo.ecommerce.orders ADD COLUMNS (discount DOUBLE)")
print("Added 'discount' column to orders table.")
spark.sql("""
INSERT INTO demo.ecommerce.orders VALUES
('ORD-0500', 'CUST-010', 'Laptop', 2, 999.99, current_timestamp(), 0.10),
('ORD-0501', 'CUST-020', 'Monitor', 1, 349.99, current_timestamp(), 0.15),
('ORD-0502', 'CUST-030', 'Keyboard', 3, 79.99, current_timestamp(), 0.05),
('ORD-0503', 'CUST-040', 'Headphones', 1, 199.99, current_timestamp(), 0.20),
('ORD-0504', 'CUST-050', 'Mouse', 4, 29.99, current_timestamp(), 0.00)
""")
print("Inserted 5 new orders with discount values.")
# Verify: old rows have NULL discount, new rows have values
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0000', 'ORD-0001', 'ORD-0500', 'ORD-0501')
ORDER BY order_id
""").show()
# =============================================================
# Cell 6: Partition Evolution (Step 3)
# =============================================================
spark.sql("""
ALTER TABLE demo.ecommerce.orders
REPLACE PARTITION FIELD days(order_date) WITH months(order_date)
""")
print("Partition strategy changed from days(order_date) to months(order_date).")
spark.sql("SELECT * FROM demo.ecommerce.orders.partitions").show(truncate=False)
# =============================================================
# Cell 7: List Snapshots (Step 4)
# =============================================================
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at
""").show(truncate=False)
# =============================================================
# Cell 8: Time Travel Query (Step 4)
# =============================================================
first_snapshot_id = spark.sql("""
SELECT snapshot_id
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at ASC
LIMIT 1
""").collect()[0][0]
print(f"First snapshot ID: {first_snapshot_id}")
# Query the table as of the first snapshot
spark.sql(f"""
SELECT count(*) AS historical_row_count
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
""").show()
# Compare with current count
spark.sql("SELECT count(*) AS current_row_count FROM demo.ecommerce.orders").show()
# Check that the old snapshot has no discount column
spark.sql(f"""
SELECT *
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
LIMIT 5
""").show()
# =============================================================
# Cell 9: View Change History (Step 4)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders.history").show(truncate=False)You can now look at any version of your table, past or present. Next up, you will use MERGE INTO to perform upserts, the real-world pattern for handling data corrections and incremental loads.
Upsert Data with MERGE INTO
In the last step, you used time travel to query your Iceberg table at previous points in time. You now have a complete audit trail of every change made to your orders table.
In production data pipelines, you often need to correct existing records and insert new ones in a single atomic operation. Iceberg supports this upsert pattern through the MERGE INTO SQL statement, the standard approach for CDC-style incremental loads.
In this step, get ready to:
- Create a corrections DataFrame with updated and new order records.
- Execute a MERGE INTO to upsert data into your Iceberg table.
Create a corrections DataFrame
Before running a MERGE INTO, you need a source dataset that contains the changes you want to apply. You will build a small DataFrame with 4 records: 2 corrections to existing orders and 2 brand new orders.
- In your Jupyter notebook, create a new cell below Cell 9 and paste the following code:
# =============================================================
# Cell 10: Create Corrections DataFrame (Step 5)
# =============================================================
# Two corrections to existing orders, two brand new orders
corrections_data = [
("ORD-0500", "CUST-010", "Laptop Pro", 2, 1299.99, datetime.now(), 0.25),
("ORD-0501", "CUST-020", "4K Monitor", 1, 449.99, datetime.now(), 0.20),
("ORD-0600", "CUST-060", "Webcam", 1, 89.99, datetime.now(), 0.00),
("ORD-0601", "CUST-070", "USB-C Hub", 2, 49.99, datetime.now(), 0.10),
]
# Define the schema explicitly to match the orders table
corrections_schema = StructType([
StructField("order_id", StringType()),
StructField("customer_id", StringType()),
StructField("product_name", StringType()),
StructField("quantity", IntegerType()),
StructField("price", DoubleType()),
StructField("order_date", TimestampType()),
StructField("discount", DoubleType()),
])
# Create the DataFrame and register it as a temporary SQL view
corrections_df = spark.createDataFrame(corrections_data, corrections_schema)
corrections_df.createOrReplaceTempView("corrections")
spark.sql("SELECT * FROM corrections").show()What does this code do?
corrections_data holds 4 tuples. The first two (ORD-0500 and ORD-0501) already exist in the orders table. Their product names and prices have been changed to simulate a correction. The last two (ORD-0600 and ORD-0601) are completely new orders.
createOrReplaceTempView("corrections") registers the DataFrame as a temporary SQL view. This lets the MERGE INTO statement in the next cell reference it by name.
- Run the cell by pressing Shift+Enter.
- You should see a table with 4 rows showing the corrections data: Laptop Pro, 4K Monitor, Webcam, and USB-C Hub.
Seeing a NameError for datetime?
The datetime module was imported in Cell 1. If you restarted the notebook kernel since then, re-run Cell 1 first to restore the import.
Help me fix a NameError for datetime in my PySpark notebook.
Execute the MERGE INTO
MERGE INTO compares each row in the source against the target table using a match condition. Rows that match get updated. Rows that don't match get inserted as new records.
- Create a new cell below Cell 10 and paste the following code:
# =============================================================
# Cell 11: MERGE INTO (Step 5)
# =============================================================
# Upsert: update matching rows, insert new rows
spark.sql("""
MERGE INTO demo.ecommerce.orders AS target
USING corrections AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
target.product_name = source.product_name,
target.price = source.price,
target.discount = source.discount
WHEN NOT MATCHED THEN INSERT *
""")
print("MERGE INTO complete.")
# Verify results
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0500', 'ORD-0501', 'ORD-0600', 'ORD-0601')
ORDER BY order_id
""").show()What does this code do?
- ON target.order_id = source.order_id is the match condition. For each row in corrections, Iceberg checks whether a row with the same order_id already exists in the orders table.
- WHEN MATCHED THEN UPDATE SET fires for ORD-0500 and ORD-0501 because those IDs already exist. It overwrites their product name, price, and discount.
- WHEN NOT MATCHED THEN INSERT * fires for ORD-0600 and ORD-0601 because those IDs are new. It inserts the entire row.
The entire operation is atomic. Either all 4 changes succeed, or none of them do.
- Run the cell by pressing Shift+Enter.
- You should see MERGE INTO complete. followed by a table showing the 4 affected rows.
- Verify that ORD-0500 now shows Laptop Pro at 1299.99, and ORD-0600 shows Webcam as a newly inserted row.
Seeing AnalysisException or table not found?
Make sure you ran Cell 10 first. The corrections temporary view only exists after that cell executes.
If you restarted your kernel, re-run all previous cells in order (Cell 1 through Cell 10) before running Cell 11.
Help me debug a MERGE INTO error in my Iceberg notebook.
Why MERGE INTO instead of separate UPDATE and INSERT?
Running separate UPDATE and INSERT statements would require two passes over the data and two separate commits. MERGE INTO does both in one atomic operation, which is faster and guarantees consistency.
This is the same pattern used in production CDC pipelines. A stream of change events arrives, and a single MERGE INTO applies all the inserts, updates, and deletes in one pass.
✔️ Awesome, I've got everything!
Great. Double check you've saved your notebook before moving on.
ⓧ I'd like to double check the full code
Here is the full lakehouse.ipynb notebook as it should look at the end of this step.
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetime
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)
# =============================================================
# Cell 5: Schema Evolution - Add Column (Step 3)
# =============================================================
spark.sql("ALTER TABLE demo.ecommerce.orders ADD COLUMNS (discount DOUBLE)")
print("Added 'discount' column to orders table.")
spark.sql("""
INSERT INTO demo.ecommerce.orders VALUES
('ORD-0500', 'CUST-010', 'Laptop', 2, 999.99, current_timestamp(), 0.10),
('ORD-0501', 'CUST-020', 'Monitor', 1, 349.99, current_timestamp(), 0.15),
('ORD-0502', 'CUST-030', 'Keyboard', 3, 79.99, current_timestamp(), 0.05),
('ORD-0503', 'CUST-040', 'Headphones', 1, 199.99, current_timestamp(), 0.20),
('ORD-0504', 'CUST-050', 'Mouse', 4, 29.99, current_timestamp(), 0.00)
""")
print("Inserted 5 new orders with discount values.")
# Verify: old rows have NULL discount, new rows have values
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0000', 'ORD-0001', 'ORD-0500', 'ORD-0501')
ORDER BY order_id
""").show()
# =============================================================
# Cell 6: Partition Evolution (Step 3)
# =============================================================
spark.sql("""
ALTER TABLE demo.ecommerce.orders
REPLACE PARTITION FIELD days(order_date) WITH months(order_date)
""")
print("Partition strategy changed from days(order_date) to months(order_date).")
spark.sql("SELECT * FROM demo.ecommerce.orders.partitions").show(truncate=False)
# =============================================================
# Cell 7: List Snapshots (Step 4)
# =============================================================
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at
""").show(truncate=False)
# =============================================================
# Cell 8: Time Travel Query (Step 4)
# =============================================================
first_snapshot_id = spark.sql("""
SELECT snapshot_id
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at ASC
LIMIT 1
""").collect()[0][0]
print(f"First snapshot ID: {first_snapshot_id}")
# Query the table as of the first snapshot
spark.sql(f"""
SELECT count(*) AS historical_row_count
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
""").show()
# Compare with current count
spark.sql("SELECT count(*) AS current_row_count FROM demo.ecommerce.orders").show()
# Check that the old snapshot has no discount column
spark.sql(f"""
SELECT *
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
LIMIT 5
""").show()
# =============================================================
# Cell 9: View Change History (Step 4)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders.history").show(truncate=False)
# =============================================================
# Cell 10: Create Corrections DataFrame (Step 5)
# =============================================================
corrections_data = [
("ORD-0500", "CUST-010", "Laptop Pro", 2, 1299.99, datetime.now(), 0.25),
("ORD-0501", "CUST-020", "4K Monitor", 1, 449.99, datetime.now(), 0.20),
("ORD-0600", "CUST-060", "Webcam", 1, 89.99, datetime.now(), 0.00),
("ORD-0601", "CUST-070", "USB-C Hub", 2, 49.99, datetime.now(), 0.10),
]
corrections_schema = StructType([
StructField("order_id", StringType()),
StructField("customer_id", StringType()),
StructField("product_name", StringType()),
StructField("quantity", IntegerType()),
StructField("price", DoubleType()),
StructField("order_date", TimestampType()),
StructField("discount", DoubleType()),
])
corrections_df = spark.createDataFrame(corrections_data, corrections_schema)
corrections_df.createOrReplaceTempView("corrections")
spark.sql("SELECT * FROM corrections").show()
# =============================================================
# Cell 11: MERGE INTO (Step 5)
# =============================================================
spark.sql("""
MERGE INTO demo.ecommerce.orders AS target
USING corrections AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
target.product_name = source.product_name,
target.price = source.price,
target.discount = source.discount
WHEN NOT MATCHED THEN INSERT *
""")
print("MERGE INTO complete.")
# Verify results
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0500', 'ORD-0501', 'ORD-0600', 'ORD-0601')
ORDER BY order_id
""").show()Your lakehouse now handles both corrections and new data in a single atomic operation. Next up, you will run production-grade maintenance operations to keep your table performant.
Maintain Your Lakehouse
Your MERGE INTO upserts are working. Your Iceberg table can now handle corrections and late-arriving data in a single atomic operation.
Every write operation you have run so far created a new snapshot. Over time, these snapshots pile up and small data files accumulate. Both waste storage and slow down queries. In this step, you will run production-grade maintenance procedures to keep your lakehouse fast and cost-efficient.
In this step, get ready to:
- Expire old snapshots to reclaim storage.
- Compact small data files into optimally-sized Parquet files.
- Run a final analytics query on the finished lakehouse.
Expire old snapshots
Iceberg creates a new snapshot for every write operation. Your table has accumulated snapshots from the initial data load, schema evolution, partition evolution, and the MERGE INTO upsert.
In production, teams schedule snapshot expiration as a recurring job. Old snapshots reference data files that can never be garbage collected until those snapshots are removed.
- Check the current state of your table before running any maintenance. Create a new cell in your notebook and add this code:
# =============================================================
# Cell 12: Check Pre-Maintenance State (Step 6)
# =============================================================
spark.sql("""
SELECT count(*) AS snapshot_count
FROM demo.ecommerce.orders.snapshots
""").show()
spark.sql("""
SELECT count(*) AS file_count
FROM demo.ecommerce.orders.files
""").show()What does this code do?
The first query counts the total number of snapshots in your table. Each snapshot is a point-in-time version created by a write operation.
The second query counts the total number of data files. Frequent writes create many small files, which slows down query performance.
- Run the cell.
- Note the snapshot_count and file_count values. You will compare these to the post-maintenance numbers shortly.
Now that you have a baseline, it is time to expire old snapshots. The expire_snapshots procedure removes old snapshots while retaining a minimum number for safety.
- Expire old snapshots by creating a new cell and adding this code:
# =============================================================
# Cell 13: Expire Snapshots (Step 6)
# =============================================================
current_ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.000")
spark.sql(f"""
CALL demo.system.expire_snapshots(
table => 'ecommerce.orders',
older_than => TIMESTAMP '{current_ts}',
retain_last => 2
)
""").show()
print("Old snapshots expired. Retained last 2.")What does this code do?
- current_ts captures the current timestamp as a formatted string. This becomes the cutoff point for expiration.
- CALL demo.system.expire_snapshots invokes Iceberg's built-in stored procedure to remove old snapshots.
- older_than sets the timestamp cutoff. Any snapshot older than this is eligible for removal.
- retain_last => 2 guarantees at least 2 snapshots survive, regardless of the timestamp cutoff. This is a safety mechanism to prevent accidental deletion of all snapshots.
- Run the cell. You should see a table showing counts of deleted data files, manifest files, and manifest lists.
Seeing an error about the procedure?
- Make sure you wrote demo.system.expire_snapshots. The procedure lives under the demo catalog's system namespace, not directly under ecommerce.
- Check that the table parameter is 'ecommerce.orders' (namespace.table) and not the full three-part name.
- Help me debug my expire_snapshots procedure call.
Compact small data files
Frequent writes to an Iceberg table create many small Parquet files. In streaming workloads, this is especially common because each micro-batch produces its own set of files.
Small files hurt query performance because the engine has to open, read, and close each one individually. Compaction merges them into fewer, larger files.
- Compact the small data files by creating a new cell and adding this code:
# =============================================================
# Cell 14: Compact Data Files (Step 6)
# =============================================================
spark.sql("""
CALL demo.system.rewrite_data_files(
table => 'ecommerce.orders'
)
""").show()
print("Data files compacted.")What does this code do?
rewrite_data_files is Iceberg's compaction procedure. It reads all the small Parquet files in your table and rewrites them into fewer, larger files. The default target file size is 512 MB.
The output table shows how many file groups were rewritten and how many files were affected. Fewer files means the query engine opens fewer file handles on every read.
- Run the cell. You should see a table showing the number of rewritten data file groups and affected files.
Compaction returned 0 rewritten groups?
This can happen if your files are already near the optimal size. Make sure you ran all the previous cells (data load, schema evolution inserts, and MERGE INTO) before compaction. Those operations create the small files that compaction merges.
Help me understand why rewrite_data_files shows 0 rewritten groups.
Run a final analytics query
Time to see the maintenance results and put your lakehouse to work. You will verify that the snapshot and file counts dropped, then run a product revenue summary to see your finished table in action.
- Verify the post-maintenance state and run a final analytics query by creating a new cell and adding this code:
# =============================================================
# Cell 15: Post-Maintenance State and Final Query (Step 6)
# =============================================================
spark.sql("""
SELECT count(*) AS snapshot_count
FROM demo.ecommerce.orders.snapshots
""").show()
spark.sql("""
SELECT count(*) AS file_count
FROM demo.ecommerce.orders.files
""").show()
# Final analytics query
spark.sql("""
SELECT
product_name,
count(*) AS total_orders,
sum(quantity) AS total_units,
round(avg(price), 2) AS avg_price,
round(sum(price * quantity), 2) AS total_revenue
FROM demo.ecommerce.orders
GROUP BY product_name
ORDER BY total_revenue DESC
""").show()What does this code do?
- The first two queries re-check the snapshot and file counts. Compare these to the numbers you recorded from Cell 12. The snapshot count should now be 2, and the file count should be lower than before.
- The final analytics query groups all orders by product_name and computes total orders, total units sold, average price, and total revenue per product.
- ORDER BY total_revenue DESC sorts the products from highest revenue to lowest. This is the kind of summary query a data analyst would run against your lakehouse.
- Run the cell.
- You should see snapshot_count as 2, a reduced file_count, and a product revenue table with 5 products (Laptop, Headphones, Keyboard, Monitor, Mouse) plus any corrected product names (Laptop Pro, 4K Monitor, Webcam, USB-C Hub).
Snapshot count is not 2?
The compaction step (Cell 14) creates a new snapshot. So after expiring and compacting, you may see 2 or 3 snapshots depending on timing. The important thing is that the count is significantly lower than before maintenance.
Help me understand my snapshot count after running expire_snapshots and rewrite_data_files.
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook before moving on.
ⓧ I'd like to double check the full code
Here is the complete lakehouse.ipynb notebook with all cells from every step.
# =============================================================
# Cell 1: Imports
# =============================================================
from pyspark.sql.functions import (
col, lit, floor, rand, concat, format_string,
current_date, current_timestamp, date_sub, expr
)
from pyspark.sql.types import (
StructType, StructField, StringType, IntegerType,
DoubleType, TimestampType
)
from datetime import datetime
# =============================================================
# Cell 2: Create Namespace and Table (Step 2)
# =============================================================
spark.sql("CREATE NAMESPACE IF NOT EXISTS demo.ecommerce")
spark.sql("""
CREATE TABLE IF NOT EXISTS demo.ecommerce.orders (
order_id STRING,
customer_id STRING,
product_name STRING,
quantity INT,
price DOUBLE,
order_date TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(order_date))
""")
print("Table demo.ecommerce.orders created successfully.")
# =============================================================
# Cell 3: Generate and Load Synthetic Data (Step 2)
# =============================================================
orders_df = (
spark.range(0, 500)
.select(
concat(lit("ORD-"), format_string("%04d", col("id"))).alias("order_id"),
concat(
lit("CUST-"),
format_string("%03d", (floor(rand() * 50) + 1).cast("int"))
).alias("customer_id"),
expr("""
CASE floor(rand() * 5)
WHEN 0 THEN 'Laptop'
WHEN 1 THEN 'Headphones'
WHEN 2 THEN 'Keyboard'
WHEN 3 THEN 'Monitor'
ELSE 'Mouse'
END
""").alias("product_name"),
(floor(rand() * 5) + 1).cast("int").alias("quantity"),
expr("round(rand() * 500 + 10, 2)").alias("price"),
expr(
"cast(date_sub(current_date(), cast(floor(rand() * 30) as int)) as timestamp)"
).alias("order_date"),
)
)
orders_df.writeTo("demo.ecommerce.orders").append()
print("Loaded 500 orders into demo.ecommerce.orders.")
# =============================================================
# Cell 4: Query and Inspect (Step 2)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders ORDER BY order_id LIMIT 10").show()
spark.sql("SELECT count(*) AS total_orders FROM demo.ecommerce.orders").show()
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
""").show(truncate=False)
# =============================================================
# Cell 5: Schema Evolution - Add Column (Step 3)
# =============================================================
spark.sql("ALTER TABLE demo.ecommerce.orders ADD COLUMNS (discount DOUBLE)")
print("Added 'discount' column to orders table.")
spark.sql("""
INSERT INTO demo.ecommerce.orders VALUES
('ORD-0500', 'CUST-010', 'Laptop', 2, 999.99, current_timestamp(), 0.10),
('ORD-0501', 'CUST-020', 'Monitor', 1, 349.99, current_timestamp(), 0.15),
('ORD-0502', 'CUST-030', 'Keyboard', 3, 79.99, current_timestamp(), 0.05),
('ORD-0503', 'CUST-040', 'Headphones', 1, 199.99, current_timestamp(), 0.20),
('ORD-0504', 'CUST-050', 'Mouse', 4, 29.99, current_timestamp(), 0.00)
""")
print("Inserted 5 new orders with discount values.")
# Verify: old rows have NULL discount, new rows have values
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0000', 'ORD-0001', 'ORD-0500', 'ORD-0501')
ORDER BY order_id
""").show()
# =============================================================
# Cell 6: Partition Evolution (Step 3)
# =============================================================
spark.sql("""
ALTER TABLE demo.ecommerce.orders
REPLACE PARTITION FIELD days(order_date) WITH months(order_date)
""")
print("Partition strategy changed from days(order_date) to months(order_date).")
spark.sql("SELECT * FROM demo.ecommerce.orders.partitions").show(truncate=False)
# =============================================================
# Cell 7: List Snapshots (Step 4)
# =============================================================
spark.sql("""
SELECT snapshot_id, committed_at, operation, summary
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at
""").show(truncate=False)
# =============================================================
# Cell 8: Time Travel Query (Step 4)
# =============================================================
first_snapshot_id = spark.sql("""
SELECT snapshot_id
FROM demo.ecommerce.orders.snapshots
ORDER BY committed_at ASC
LIMIT 1
""").collect()[0][0]
print(f"First snapshot ID: {first_snapshot_id}")
# Query the table as of the first snapshot
spark.sql(f"""
SELECT count(*) AS historical_row_count
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
""").show()
# Compare with current count
spark.sql("SELECT count(*) AS current_row_count FROM demo.ecommerce.orders").show()
# Check that the old snapshot has no discount column
spark.sql(f"""
SELECT *
FROM demo.ecommerce.orders
VERSION AS OF {first_snapshot_id}
LIMIT 5
""").show()
# =============================================================
# Cell 9: View Change History (Step 4)
# =============================================================
spark.sql("SELECT * FROM demo.ecommerce.orders.history").show(truncate=False)
# =============================================================
# Cell 10: Create Corrections DataFrame (Step 5)
# =============================================================
corrections_data = [
("ORD-0500", "CUST-010", "Laptop Pro", 2, 1299.99, datetime.now(), 0.25),
("ORD-0501", "CUST-020", "4K Monitor", 1, 449.99, datetime.now(), 0.20),
("ORD-0600", "CUST-060", "Webcam", 1, 89.99, datetime.now(), 0.00),
("ORD-0601", "CUST-070", "USB-C Hub", 2, 49.99, datetime.now(), 0.10),
]
corrections_schema = StructType([
StructField("order_id", StringType()),
StructField("customer_id", StringType()),
StructField("product_name", StringType()),
StructField("quantity", IntegerType()),
StructField("price", DoubleType()),
StructField("order_date", TimestampType()),
StructField("discount", DoubleType()),
])
corrections_df = spark.createDataFrame(corrections_data, corrections_schema)
corrections_df.createOrReplaceTempView("corrections")
spark.sql("SELECT * FROM corrections").show()
# =============================================================
# Cell 11: MERGE INTO (Step 5)
# =============================================================
spark.sql("""
MERGE INTO demo.ecommerce.orders AS target
USING corrections AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
target.product_name = source.product_name,
target.price = source.price,
target.discount = source.discount
WHEN NOT MATCHED THEN INSERT *
""")
print("MERGE INTO complete.")
# Verify results
spark.sql("""
SELECT order_id, product_name, price, discount
FROM demo.ecommerce.orders
WHERE order_id IN ('ORD-0500', 'ORD-0501', 'ORD-0600', 'ORD-0601')
ORDER BY order_id
""").show()
# =============================================================
# Cell 12: Check Pre-Maintenance State (Step 6)
# =============================================================
spark.sql("""
SELECT count(*) AS snapshot_count
FROM demo.ecommerce.orders.snapshots
""").show()
spark.sql("""
SELECT count(*) AS file_count
FROM demo.ecommerce.orders.files
""").show()
# =============================================================
# Cell 13: Expire Snapshots (Step 6)
# =============================================================
current_ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.000")
spark.sql(f"""
CALL demo.system.expire_snapshots(
table => 'ecommerce.orders',
older_than => TIMESTAMP '{current_ts}',
retain_last => 2
)
""").show()
print("Old snapshots expired. Retained last 2.")
# =============================================================
# Cell 14: Compact Data Files (Step 6)
# =============================================================
spark.sql("""
CALL demo.system.rewrite_data_files(
table => 'ecommerce.orders'
)
""").show()
print("Data files compacted.")
# =============================================================
# Cell 15: Post-Maintenance State and Final Query (Step 6)
# =============================================================
spark.sql("""
SELECT count(*) AS snapshot_count
FROM demo.ecommerce.orders.snapshots
""").show()
spark.sql("""
SELECT count(*) AS file_count
FROM demo.ecommerce.orders.files
""").show()
# Final analytics query
spark.sql("""
SELECT
product_name,
count(*) AS total_orders,
sum(quantity) AS total_units,
round(avg(price), 2) AS avg_price,
round(sum(price * quantity), 2) AS total_revenue
FROM demo.ecommerce.orders
GROUP BY product_name
ORDER BY total_revenue DESC
""").show()Secret mission
Build a Medallion Architecture
Your lakehouse can create tables, evolve schemas, and time travel. But production data teams don't store everything in a single table. Can you build a Bronze/Silver/Gold medallion architecture on top of your Iceberg lakehouse?
Before you go
Clean Up Your Resources
This project runs entirely locally using Docker containers, so there are no ongoing cloud charges.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- 4 Docker containers: spark-iceberg, iceberg-rest, minio, and mc.
- Local Docker volumes for warehouse data and notebooks.
- The docker-compose.yml file and lakehouse.ipynb notebook in your iceberg-lakehouse folder.
Keep everything running
No action needed. Choose this if you are still actively experimenting with your lakehouse.
- Your Jupyter notebook and all Iceberg tables persist across container restarts.
- Access Jupyter at http://localhost:8888 and MinIO at http://localhost:9001 any time the containers are running.
- The containers use minimal resources while idle, so there is no cost to leaving them up.
Pause - I'll come back to this later
Shut down all running containers to free up memory and CPU without losing any data.
- Open a terminal in your iceberg-lakehouse folder.
- Stop all containers by running:
docker-compose stop- When you are ready to resume, start the containers again by running:
docker-compose startAll your Iceberg tables, notebook code, and MinIO data will be exactly where you left them.
Delete - I don't want to use this again
Remove all containers and volumes to start completely fresh. This permanently deletes all Iceberg tables and data.
- Open a terminal in your iceberg-lakehouse folder.
- Stop and remove all containers and volumes by running:
docker-compose down -v- Optionally, delete the iceberg-lakehouse folder from your Desktop if you no longer need the docker-compose.yml or notebook files.
To rebuild the project from scratch, follow the steps in this project again starting from Step 1.
Mission accomplished
That's a Wrap!
Nice work! You just built a modern data lakehouse from scratch using Apache Iceberg, PySpark, and MinIO. These are the exact tools and patterns that data engineering teams at companies like Netflix, Apple, and LinkedIn rely on every day.
You've learned how to:
- Built a complete lakehouse environment with PySpark, Apache Iceberg, and MinIO, all orchestrated by Docker Compose.
- Evolved your table's schema and partitioning strategy without rewriting a single data file, then queried historical snapshots using time travel.
- Performed atomic MERGE INTO upserts for CDC-style incremental loads, then ran production table maintenance operations like snapshot expiration and data file compaction.
- Secret Mission: Built a Bronze/Silver/Gold medallion architecture on top of your lakehouse.
Ready to quiz yourself?