6.18. join_optimization-tpch-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
from pyspark.sql import functions as F

spark = SparkSession.builder.remote("sc://spark-connect:15002").getOrCreate()

Both setup implementations choose an explicit notebook scale override before DATASET_SCALE, default to small, and call the internal resolver once. They validate and retain one verified immutable generation as datasetObjectByName / dataset_object_by_name.

2.2 Read¶

Scala (Zeppelin):

val orders = spark.read.parquet(datasetObjectByName("orders.parquet"))
val customer = spark.read.parquet(datasetObjectByName("customer.parquet"))

PySpark (Jupyter):

orders = spark.read.parquet(dataset_object_by_name["orders.parquet"])
customer = spark.read.parquet(dataset_object_by_name["customer.parquet"])

2.3 Transform¶

Scala (Zeppelin):

val joined = orders.join(broadcast(customer), $"o_custkey" === $"c_custkey")
joined.explain()
val mart = joined.groupBy($"c_mktsegment").agg(sum($"o_totalprice").as("revenue"), count("*").as("orders"))

PySpark (Jupyter):

joined = orders.join(F.broadcast(customer), F.col("o_custkey") == F.col("c_custkey"))
joined.explain()
mart = joined.groupBy("c_mktsegment").agg(F.sum("o_totalprice").alias("revenue"), F.count("*").alias("orders"))

2.4 Write¶

Scala (Zeppelin):

mart.writeTo("lakehouse.gold.tpch_segment_revenue").using("iceberg").createOrReplace()

PySpark (Jupyter):

mart.writeTo("lakehouse.gold.tpch_segment_revenue").using("iceberg").createOrReplace()

2.5 Verify¶

Scala (Zeppelin):

println("AQE: " + spark.conf.get("spark.sql.adaptive.enabled"))
spark.table("lakehouse.gold.tpch_segment_revenue").orderBy($"revenue".desc).show(false)

PySpark (Jupyter):

print("AQE:", spark.conf.get("spark.sql.adaptive.enabled"))
spark.table("lakehouse.gold.tpch_segment_revenue").orderBy(F.desc("revenue")).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.