6.15. incremental_upsert-online_retail-spark-iceberg¶

Documents the scenario's paired Jupyter (notebook.ipynb) and Zeppelin (notebook.zpln) implementations. Both notebooks implement identical logic in PySpark and Scala.

1. Section map¶

Subsection	Scala (Zeppelin)	PySpark (Jupyter)
2.1 Setup	✓	✓
2.2 Read	✓	✓
2.3 Transform	✓	✓
2.4 Write	✓	✓
2.5 Verify	✓	✓

2. Walkthrough¶

2.1 Setup¶

Scala (Zeppelin):

import spark.implicits._
import org.apache.spark.sql.functions._
// spark pre-bound (Spark Connect + lakehouse catalog)

PySpark (Jupyter):

from pyspark.sql import SparkSession

spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()
# lakehouse catalog pre-configured

2.2 Read¶

Scala (Zeppelin):

spark.sql("CREATE TABLE IF NOT EXISTS lakehouse.silver.online_retail (invoice string, stock_code string, quantity int, price double) USING iceberg").show(false)
spark.sql("INSERT INTO lakehouse.silver.online_retail VALUES ('A1','SKU1',5,2.0), ('A2','SKU2',3,4.0)").show(false)
spark.sql("SELECT * FROM lakehouse.silver.online_retail ORDER BY invoice").show(false)

PySpark (Jupyter):

spark.sql("CREATE TABLE IF NOT EXISTS lakehouse.silver.online_retail (invoice string, stock_code string, quantity int, price double) USING iceberg").show(truncate=False)
spark.sql("INSERT INTO lakehouse.silver.online_retail VALUES ('A1','SKU1',5,2.0), ('A2','SKU2',3,4.0)").show(truncate=False)
spark.sql("SELECT * FROM lakehouse.silver.online_retail ORDER BY invoice").show(truncate=False)

2.3 Transform¶

Scala (Zeppelin):

spark.sql("CREATE OR REPLACE TEMP VIEW online_retail_updates AS SELECT * FROM VALUES ('A1','SKU1',10,2.0), ('A3','SKU3',1,9.0) AS t(invoice, stock_code, quantity, price)").show(false)

PySpark (Jupyter):

spark.sql("CREATE OR REPLACE TEMP VIEW online_retail_updates AS SELECT * FROM VALUES ('A1','SKU1',10,2.0), ('A3','SKU3',1,9.0) AS t(invoice, stock_code, quantity, price)").show(truncate=False)

2.4 Write¶

Scala (Zeppelin):

spark.sql("MERGE INTO lakehouse.silver.online_retail t USING online_retail_updates s ON t.invoice = s.invoice AND t.stock_code = s.stock_code WHEN MATCHED THEN UPDATE SET t.quantity = s.quantity WHEN NOT MATCHED THEN INSERT *").show(false)

PySpark (Jupyter):

spark.sql("MERGE INTO lakehouse.silver.online_retail t USING online_retail_updates s ON t.invoice = s.invoice AND t.stock_code = s.stock_code WHEN MATCHED THEN UPDATE SET t.quantity = s.quantity WHEN NOT MATCHED THEN INSERT *").show(truncate=False)

2.5 Verify¶

Scala (Zeppelin):

spark.sql("SELECT * FROM lakehouse.silver.online_retail ORDER BY invoice").show(false)
// Re-running the MERGE is idempotent: same result.

PySpark (Jupyter):

spark.sql("SELECT * FROM lakehouse.silver.online_retail ORDER BY invoice").show(truncate=False)
print("Re-running the MERGE is idempotent: same result.")

3. Scala / PySpark parity¶

Both notebooks share the same numbered sections and produce identical Iceberg tables; only the language and interpreter differ.

4. How to run¶

Open the scenario's zeppelin/notebook.zpln on the Atlas Zeppelin UI or jupyter/notebook.ipynb on JupyterHub, then run all paragraphs/cells top to bottom.