6.11. feature_engineering-movielens-spark-iceberg¶

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

Production trust boundary: these notebooks are educational parity surfaces, not a supported production write path. They infer schema and directly replace the same tables without complete-publication validation, production provenance, Airflow serialization, or readback checks. Running them against production tables can invalidate downstream provenance. Use movielens_feature_pipeline from spark-apps/movielens-feature-pipeline for production writes.

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 ratings = spark.read.option("header", true).option("inferSchema", true).csv(datasetObjectByName("ratings.csv"))

PySpark (Jupyter):

ratings = spark.read.option("header", True).option("inferSchema", True).csv(dataset_object_by_name["ratings.csv"])

2.3 Transform¶

Scala (Zeppelin):

val userFeatures = ratings.groupBy($"userId").agg(avg($"rating").as("avg_rating"), count("*").as("num_ratings"))
val movieFeatures = ratings.groupBy($"movieId").agg(avg($"rating").as("movie_avg"), count("*").as("popularity"))

PySpark (Jupyter):

userFeatures = ratings.groupBy("userId").agg(F.avg("rating").alias("avg_rating"), F.count("*").alias("num_ratings"))
movieFeatures = ratings.groupBy("movieId").agg(F.avg("rating").alias("movie_avg"), F.count("*").alias("popularity"))

Duplicate rating rows intentionally count separately: each input row contributes once to the relevant average and count(*).

2.4 Write¶

Scala (Zeppelin):

userFeatures.writeTo("lakehouse.gold.ml_user_features").using("iceberg").createOrReplace()
movieFeatures.writeTo("lakehouse.gold.ml_movie_features").using("iceberg").createOrReplace()

PySpark (Jupyter):

userFeatures.writeTo("lakehouse.gold.ml_user_features").using("iceberg").createOrReplace()
movieFeatures.writeTo("lakehouse.gold.ml_movie_features").using("iceberg").createOrReplace()

2.5 Verify¶

Scala (Zeppelin):

spark.table("lakehouse.gold.ml_movie_features").orderBy($"popularity".desc).show(10, false)

PySpark (Jupyter):

spark.table("lakehouse.gold.ml_movie_features").orderBy(F.desc("popularity")).show(10, 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.