This project is a batch recommendation pipeline built around MovieLens data. Spark handles CSV ingestion, Parquet tables, similarity generation, ALS training, and exporting the output to a small database. The API reads the database produced by the Spark jobs and the web app calls the API. The web app never talks to Spark or reads Parquet directly.

The source dataset is Movielens 25M. It contains 25 million ratings, movie metadata, user tags, and 15.5 million tag genome data. The ratings are MovieLens user ratings, not IMDb scores. The tag genome data is what makes this project more interesting. Each movie has relevance scores for tags like crime, dark comedy, tense, etc., which gives us enough data for a content-based recommendation path alongside collaborative filtering.

Architecture

The system has the following storage and boundaries:

MovieLens CSV
  -> bronze Parquet
  -> silver domain tables
  -> gold recommendation tables
  -> SQLite serving database
  -> API
  -> Web App

Parquet layers

The Spark jobs write a local Parquet lake split into bronze, silver, and gold layers. The tables are just plain Parquet directories managed by the jobs.

Bronze

Bronze is the raw source data converted to Parquet with explicit schemas. It does minimal transformation. The ingestion job reads the MovieLens CSV files and writes Parquet tables to the bronze layer (under lake/bronze).

The source schemas are part of the contract. The job should fail if a required column disappears or changes type.

RATINGS_SCHEMA = StructType(
    [
        StructField("userId", IntegerType(), False),
        StructField("movieId", IntegerType(), False),
        StructField("rating", DoubleType(), False),
        StructField("timestamp", LongType(), False),
    ]
)

The bronze layer exists mostly to avoid reading the CSV file repeatedly and to establish the source schema.

Example:

// ratings.csv

userId,movieId,rating,timestamp
1,296,5.0,1147880044
1,306,3.5,1147868817

becomes

// bronze.ratings

userId    movieId    rating    timestamp
1         296        5.0       1147880044
1         306        3.5       1147868817

Silver

Silver is where raw MovieLens data becomes useful tables. It normalizes dates, titles, genres, and the feature tables required by the recommendation jobs.

Silver tables:

lake/silver/
  movies/
  ratings/
  tags/
  movie_genres/
  movie_tag_profiles/
  user_genre_profiles/
  user_tag_profiles/

The job separates release year from the title, joins IMDb and TMDb identifiers, and turns the pipe-delimited genre fields into an array. Additionally, the separate jobs normalize rating timestamps and explode genres into one row per movie and genre.

return (
    movies
    .join(links, "movieId", "left")
    .select(
        F.col("movieId").alias("movie_id"),
        F.col("title"),
        normalized_title.alias("normalized_title"),
        F.when(title_year != "", title_year.cast("int")).alias("release_year"),
        F.split(F.col("genres"), r"|").alias("genres"),
        F.col("imdbId").alias("imdb_id"),
        F.col("tmdbId").alias("tmdb_id"),
    )
)

The rating >= 4.0 filter treats high ratings as positive evidence. It should eventually be evaluated against other thresholds in the future.

Once the domain tables are stable, the gold jobs build ranked and exportable outputs.

Gold

Gold is the output layer. This is where Spark creates outputs and the tables that can be served, exported, and used by the SQLite export.

lake/gold/
  movie_metrics/
  public_movies/
  top_movie_tags/
  similar_movies/
  als_user_recommendations/
  als_movie_factors/
  als_user_factors/

Movie-level statistics (such as rating count, average rating, and Bayesian rating) are also calculated here.

For ratings, a Bayesian-style score is used instead of the raw average because small sample sizes can distort the ratings.

bayesian_rating = ((v x R) + (m x C)) / (v + m)

R = the movie's average rating
v = the movie's rating count
C = the global average rating
m = the prior weight

The current job uses a prior weight of 50. This controls how quickly a movie's own average dominates the global prior.

Movie similarity

The similarity job uses the tag genome data by converting each movie to a vector. We then use cosine similarity to compare tag profiles. The job does not cross-join every movie with every other movie. A cross join over 5,000 movies creates 25 million candidate rows before scoring. Instead, both sides are joined on tag_id, so a pair is created only when the movies share a retained tag. The final table keeps the top 24 results per source movie. With 5,000 movies and 24 results per movie, the completed job produces 120,000 similarity rows.

The second recommendation path uses Spark ML ALS. ALS factorizes the user-movie rating matrix into two lower-dimensional matrices (one vector per user and one vector per movie). A predicted rating is the dot product of these two vectors. The Spark Job precomputes recommendations and writes ranked rows to gold.als_user_recommendations.

als = ALS(
  userCol="user_id",
  itemCol="movie_id",
  ratingCol="rating",
  rank=32,
  maxIter=10,
  regParam=0.08,
  implicitPrefs=False,
  coldStartStrategy="drop",
  nonnegative=True,
)

model = als.fit(training_ratings)

Neither similarity generation nor ALS runs during an API request. Both outputs are ready before the serving database is built.

SQLite DB

The SQLite DB is the final small read model used by the API. It contains the 5,000 movies, the FTS search index and 120,000 precomputed similarity rows. The 25 million ratings and 15.5 million genome scores remain in Parquet.

Text search uses SQLite FTS table movie_search

db.prepare(`
  SELECT m.*
  FROM movie_search s
  JOIN movies m ON m.movie_id = s.rowid
  WHERE movie_search MATCH ?
  ORDER BY bm25(movie_search), m.rating_count DESC
  LIMIT ?
`)

Similar movies are served via an indexed lookup against the exported similar_movies table.

Runtime

The application runs in three containers.

• dev container: runs Spark jobs and writes Parquet and SQLite artifacts.
• api container: reads SQLite serving artifact.
• web container: serves the static frontend and proxies /api

Next steps

The current version does not evaluate recommendation quality. RMSE can measure rating-prediction error, but a lower RMSE does not necessarily mean that the top ten recommendations are better. For this system, ranking metrics such as precision@K, recall@K, MAP@K, and NDCG@K are more useful. A better evaluation would train on earlier ratings and measures whether movies rated positively later are ranked near the top.

The next step is measuring which candidates are actually useful, versioning the model and dataset together, and making the exported results reproducible across runs.