6.12. scd2-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.gold.dim_customer_scd2 (customer_id string, segment string, effective_from timestamp, effective_to timestamp, is_current boolean) USING iceberg").show(false)
spark.sql("INSERT INTO lakehouse.gold.dim_customer_scd2 VALUES ('C1','standard', TIMESTAMP '2023-01-01 00:00:00', NULL, true)").show(false)
spark.sql("SELECT customer_id, segment, is_current FROM lakehouse.gold.dim_customer_scd2 ORDER BY effective_from").show(false)

PySpark (Jupyter):

spark.sql("CREATE TABLE IF NOT EXISTS lakehouse.gold.dim_customer_scd2 (customer_id string, segment string, effective_from timestamp, effective_to timestamp, is_current boolean) USING iceberg").show(truncate=False)
spark.sql("INSERT INTO lakehouse.gold.dim_customer_scd2 VALUES ('C1','standard', TIMESTAMP '2023-01-01 00:00:00', NULL, true)").show(truncate=False)
spark.sql("SELECT customer_id, segment, is_current FROM lakehouse.gold.dim_customer_scd2 ORDER BY effective_from").show(truncate=False)

2.3 Transform

Scala (Zeppelin):

spark.sql("UPDATE lakehouse.gold.dim_customer_scd2 SET effective_to = current_timestamp(), is_current = false WHERE customer_id = 'C1' AND is_current = true").show(false)

PySpark (Jupyter):

spark.sql("UPDATE lakehouse.gold.dim_customer_scd2 SET effective_to = current_timestamp(), is_current = false WHERE customer_id = 'C1' AND is_current = true").show(truncate=False)

2.4 Write

Scala (Zeppelin):

spark.sql("INSERT INTO lakehouse.gold.dim_customer_scd2 VALUES ('C1','premium', current_timestamp(), NULL, true)").show(false)

PySpark (Jupyter):

spark.sql("INSERT INTO lakehouse.gold.dim_customer_scd2 VALUES ('C1','premium', current_timestamp(), NULL, true)").show(truncate=False)

2.5 Verify

Scala (Zeppelin):

spark.sql("SELECT customer_id, segment, effective_to IS NOT NULL AS closed, is_current FROM lakehouse.gold.dim_customer_scd2 ORDER BY effective_from").show(false)

PySpark (Jupyter):

spark.sql("SELECT customer_id, segment, effective_to IS NOT NULL AS closed, is_current FROM lakehouse.gold.dim_customer_scd2 ORDER BY effective_from").show(truncate=False)

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.