CineLake is a project built around MovieLens data. It is based around Apache Spark, that handles CSV ingestion, Parquet tables, and similarity generation. The similarity and training data is saved to a database and exposed through APIs. A small web app then calls these APIs.

The dataset contains 25 million movie ratings, metadata, user tags, and 15.5 million tag gnome data. It is important to note that the ratings are MovieLens user ratings, not IMDb scores. Each movie has relevance scores for tags like comedy, thriller, crime etc. which gives us enough data for a recommendation pipeline.

Architecture

Apache Spark is a distributed engine for large-scale data processing. At a high level, Spark takes our operations and executes them efficiently.

For example - movie_ratings.groupBy(“movie_id”).agg(F.avg(“rating”).alias(“avg_rating”) means take ratings, group them by movie and calculate average ratings. This is done by dividing work across multiple cores.

Sure, we could use Pandas here, but Spark has built-in functionality for large joins, aggregation, Parquet, machine-learning and distributed execution. Also, the plan here is to build a pipeline that can scale beyond MovieLens without having to redesign the entire model.

Our project has the following structure:

Spark Jobs

Spark jobs are just scripts – ingest, build_silver, build_similarity, export_sqlite etc.

Parquet layers

Parquet is a file format for storing analytical data. The main benefit is it can read the column that it needs without having to read every field for every row. Additionally, it supports schemas, compression, nested structures, metadata and more. Finally, Spark has excellent Parquet support.
Why not use CSV? Because it is not a good analytical storage format. With CSV, every file read means opening file, parsing separators and strings, converting data to double or integer etc. With parquet, ingestion happens once, with resulting data already having a schema.

We are following the medallion architecture pattern. Bronze, Silver and gold are progressive versions of the data. These are not Spark features, these are just separating out reponsibilities. The basic idea is bronze is raw data, silver is cleaned data and gold is proper output.

Bronze

Bronze layer simply adds type, validation and compression to the data. It does very minimal transformation, and exists mostly to avoid reading the CSV file again and again.

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

Example:

// ratings.csv

userId,movieId,rating,timestamp
1,206,5.0,1258452122
1,396,3.6,1258452156

// bronze.ratings

userId    movieId    rating    timestamp
1         206        5.0       1258452122
1         396        3.6       1258452156

Silver

The silver layer converts raw data into domain data. It normalizes dates, titles, genres, and tags.

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

We also nomarlize timestamps, join IMDb and TMDB ids and explode genres into one row per movie and genre here.

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"),
    )
)

Gold

Gold is the output layer. This is where our data gets ready to be used by the SQLite export.

// Gold tables

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

Movie similarity

The first recommendation path is content based. The question to ask is if I like this movie, what are other movies that have similar characteristics? This is where the tag genome is useful. Each movie has relevance scores across tags.

For example:

Movie A: [0.85, 0.45, 0.05, 0.80]
Movie B: [0.82, 0.30, 0.10, 0.78]
Movie C: [0.05, 0.95, 0.96, 0.09]

Movie and Movie B are much more similar.

We can compare these vectors using cosine similarity.

cosine(A, B) = A.B/||A||x||B||

The numerator measures the overlap, and denominator normalizes it for vector magnitude. The higher scores indicate more similarities here. We also do not need to keep very possible related movie. Keeping 24 recommendations per movie means 5,000 * 24 = 120,000 similarity rows.

movie_id | similar_movie_id | score | rank
1        | 20               | 0.950 |  1   
2        | 250              | 0.943 |  2   
…

The second recommendation path is ALS (Alternating Least Squares). It factorizes the matrix into two matrices (one per user and one per movie). Both user and movie receive a latent vector each and predicted rating is a dot product of those.

predicted_rating = user_vector x movie_vector

Our Spark job precomputes these and writes ranked rows to gold.als_user_recommendations.
Again, both similarity generation and ALS outputs are ready before building the serving database.

SQLite DB

After the gold jobs complete, we write a SQLite database. It is a serving database, containing information only required by the API. This database contains:

• 5,000 movies
• 120,000 similarity data
• search index
• movie statistics

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

This project generates recommendations, but it does not evaluate their quality. It is easy to output that movie B is a recommendation for movie A, but how do we tell movie B is a useful recommendation?

RMSE (Root Mean Squared Error) is one way to evaluate ALS predictions. It tells me how far the rating prediction is from the ratings given by users. A lower RMSE means more accurate predictions. But it also means we are predicting how a user would rate a certain movie not what 3 movies should the user watch next?

For that, I need retrieval metrics such as NDCG and MAP. These metrics deal with continuous relevance rather than just a strict binary (relevant or not relevant) score. These are closer to what I actually want to measure, such as how many of the recommended movies are actually relevant, or did the best recommendation appear at the top?