Rutuja Jadhav

1. Data Preprocessing

1.1. Preliminary Inspection

Here is a background about the files obtained from Group Lens.

ML Small Dataset from Group Lens: 100,000 ratings and 3,600 tag applications applied to 9,000 movies by 600 users. Last updated 9/2018.

User Ids

MovieLens users were selected at random for inclusion. Their ids have been anonymized. User ids are consistent between ratings.csv and tags.csv (i.e., the same id refers to the same user across the two files).

Movie Ids

Only movies with at least one rating or tag are included in the dataset. These movie ids are consistent with those used on the MovieLens web site (e.g., id 1 corresponds to the URL https://movielens.org/movies/1). Movie ids are consistent between ratings.csv, tags.csv, movies.csv, and links.csv (i.e., the same id refers to the same movie across these four data files).

Ratings Data File Structure (ratings.csv)

All ratings are contained in the file ratings.csv. Each line of this file after the header row represents one rating of one movie by one user, and has the following format:

userId,movieId,rating,timestamp

The lines within this file are ordered first by userId, then, within user, by movieId.

Ratings are made on a 5-star scale, with half-star increments (0.5 stars - 5.0 stars).

Timestamps represent seconds since midnight Coordinated Universal Time (UTC) of January 1, 1970.

Ratings Table

userId	movieId	rating	timestamp
1	1	4	964982703
1	3	4	964981247
1	6	4	964982224
1	47	5	964983815
1	50	5	964982931
1	70	3	964982400
1	101	5	964980868
1	110	4	964982176
1	151	5	964984041
1	157	5	964984100

Tags Data File Structure (tags.csv)

All tags are contained in the file tags.csv. Each line of this file after the header row represents one tag applied to one movie by one user, and has the following format:

userId,movieId,tag,timestamp

The lines within this file are ordered first by userId, then, within user, by movieId.

Tags are user-generated metadata about movies. Each tag is typically a single word or short phrase. The meaning, value, and purpose of a particular tag is determined by each user.

Timestamps represent seconds since midnight Coordinated Universal Time (UTC) of January 1, 1970.

Movies Data File Structure (movies.csv)

Movie information is contained in the file movies.csv. Each line of this file after the header row represents one movie, and has the following format:

movieId,title,genres

Movie titles are entered manually or imported from https://www.themoviedb.org/, and include the year of release in parentheses. Errors and inconsistencies may exist in these titles.

Genres are a pipe-separated list, and are selected from the following:

Movies Table

movieId	title	genres
1	Toy Story (1995)	Adventure\|Animation\|Children\|Comedy\|Fantasy
2	Jumanji (1995)	Adventure\|Children\|Fantasy
3	Grumpier Old Men (1995)	Comedy\|Romance
4	Waiting to Exhale (1995)	Comedy\|Drama\|Romance
5	Father of the Bride Part II (1995)	Comedy
6	Heat (1995)	Action\|Crime\|Thriller
7	Sabrina (1995)	Comedy\|Romance
8	Tom and Huck (1995)	Adventure\|Children
9	Sudden Death (1995)	Action
10	GoldenEye (1995)	Action\|Adventure\|Thriller

Links Data File Structure (links.csv)

Identifiers that can be used to link to other sources of movie data are contained in the file links.csv. Each line of this file after the header row represents one movie, and has the following format:

movieId,imdbId,tmdbId

movieId is an identifier for movies used by https://movielens.org. E.g., the movie Toy Story has the link https://movielens.org/movies/1.

imdbId is an identifier for movies used by http://www.imdb.com. E.g., the movie Toy Story has the link http://www.imdb.com/title/tt0114709/.

tmdbId is an identifier for movies used by https://www.themoviedb.org. E.g., the movie Toy Story has the link https://www.themoviedb.org/movie/862.

Use of the resources listed above is subject to the terms of each provider.

Preliminary Observation Summary

Users: 610

Movies: 9,724

Ratings (observed interactions): 100,836

Total Possible interactions: 5,931,640

Sparsity: 98.30%

The main table for ALS is the ratings table (users and their movie ratings) which recorded 100,836 interactions out of a possible 5.9M revealing 98.3& sparsity! The overwhelming majority of possible user-movie interactions are therefore unobserved. This sparsity signals that recommender would have to learn from the smaller subset of observed interactions via methods such a collaborative filtering and ALS.

File formats - csv vs parquet

At NYU, working on the HPC cluster with genuinely large datasets, we always converted raw data to Parquet before doing any real work — it wasn’t optional, since re-parsing large CSVs on every job run was slow enough to notice and annoying enough to avoid. So going into this project, I half-expected to do the same thing here out of habit.

For this project, working with the ml-latest-small dataset (100K rows, ~1MB), I read directly from CSV throughout and never bothered converting to Parquet. Practically it made zero difference — the dataset is small enough that Spark chews through it in seconds regardless of format. And this actually lines up with something we tested in my Big Data class at NYU: Parquet can defeat its own purpose on smaller datasets, because the conversion itself — schema inference, columnar encoding, compression, writing the file back out — has a real cost, and that cost only pays for itself once you’re reading the resulting file many separate times afterward. On a single pass, you’re just paying the overhead with nothing to amortize it against.

I re-ran that exact experiment here to check it still held:

CSV read: 1.738s

Parquet write (once): 2.179s

Parquet read: 1.033s

Parquet total (write+read): 3.212s

Parquet’s read really is faster than CSV’s (1.033s vs 1.738s), but the one-time write cost (2.179s) more than eats up that savings, so total time-to-first-use is nearly double CSV’s. And in this project specifically, everything downstream (train, val, test, model fitting) operates on DataFrames derived once from that first read, sitting in Spark’s lazy execution plan — not repeated fresh reads off disk — so there’s no “many reads” happening to amortize the conversion against in the first place.

If I scaled this project up to ml-25m (as I discuss in the scaling section), converting to Parquet early would be one of the first things I’d add back in, especially if a pipeline is re-reading the same source repeatedly across independent runs rather than once per notebook session.

1.2. Data Splitting Strategy

Before touching any model, I needed to figure out how to split ratings into train/val/test — and this turned out to be less trivial than a standard random split, because of something specific to Rec-Sys as opposed to other ML pipelines.

Potential problems with random split

Problem I - Blindspots

The problem with a plain random split: ALS (and other rec-sys models like popularity baselines, similarity, etc) doesn’t work like a typical regression model that generalizes from features across all users. Instead, it learns one dedicated latent vector per user and per movie, directly from that entity’s own interaction history. If a user’s ratings all happened to land in train by chance (or worse, all in test), there’d be no way to solve for their latent vector at all.

Scenario	What breaks	Why
User’s ratings all in test	Training	No data to solve `u_i` from — genuinely can’t learn a meaningful vector
User’s ratings all in train	Evaluation	Vector gets learned fine, but there’s no held-out truth to check it against — that user silently disappears from your reported metrics

The most closest other ML analogy I can think of is a class imbalance problem in classification. If the random split assigns labels of the minority class only to the test set, the train set will not contain anything to learn how to predict the minority class. Likewise if every class lable of some class class K ends up in train, the pipeline will not have any data held out in val or test to tune or evaluate the model with.

Problem II- Temporal data

A random split would also disturb the temporal nature of rec-sys data, making it quite unrealistic. People’s preferences evolve over time naturally.

Movie-Lens split solution

I confirmed the dataset’s minimum was 20 ratings/user, which meant a careful split could safely guarantee every user shows up in every split. A pure random split couldn’t guarantee that.

min_ratings	max_ratings	avg_ratings	median_ratings
20	2698	165.30491803278687	70

The fix I used : Per-user, chronological splitting.

For each user individually, sort their ratings by timestamp, then reserve their most recent ~10% for test, the next ~10% for val, and the rest (earliest ~80%) for train. This does two things at once:

Guarantees every user has some data in every split (given the 20-rating minimum)
Makes the evaluation realistic — predicting a user’s future ratings from their past ratings, rather than randomly hiding ratings from the middle of their timeline

Key Spark Operations Used Here

Window.partitionBy("userId").orderBy("timestamp")

Defines a per-user, time-sorted window. This is the backbone of the whole approach. It lets us rank and count things within each user’s own group, without collapsing the individual rows the way a GROUP BY would.
F.row_number().over(w)

Assigns 1, 2, 3, ... to each user’s ratings in chronological order, restarting at 1 for every user because of the partitionBy. This becomes each rating’s position within that user’s history.
F.count("*").over(Window.partitionBy("userId"))

Computes a group-level total — in this case, how many ratings the user has in total — and stamps that number onto every row belonging to that user. Unlike GROUP BY, the individual rating rows are preserved.
F.ceil(...)

Used when computing each user’s cutoff size: user_total * 0.10 Rounding up is a small but important detail because it guarantees that even a user with only 20 ratings gets at least 1 rating reserved for validation and 1 for test. Without ceil(), a low-count user could round down to zero, silently breaking the guarantee that every user has data in every split.
F.when(...).otherwise(...)

Provides Spark’s if/else-style column logic. Here, it is used to label each rating as train, val, or test based on comparing its row_number against the user’s total number of ratings and the calculated cutoffs.
GROUP BY vs. PARTITION BY

It is tempting to think these two approaches do the same thing, but they don’t. A GROUP BY collapses multiple rows into one summary row per group. For example, I used GROUP BY earlier for EDA when counting the number of ratings per user. By contrast, PARTITION BY inside a window function keeps every original row. It simply computes something relative to that row’s group and attaches the result back to the original data. This distinction is important for the train/validation/test split. I needed every individual rating to survive into one of the three datasets, rather than reducing each user to a single summary row. Therefore, the window-function approach using PARTITION BY was the appropriate choice.

Split Result

As a final sanity check, I counted the number of distinct users appearing in each split:

train_users = train.select("userId").distinct().count()

print(f"Users in train: {train_users} (should equal {n_users})")

I ran the same check for validation and test.

The results confirmed that all 610 users appeared in all three sets:

Split	Number of Users
Train	610
Validation	610
Test	610

This confirms that the user-level splitting logic successfully preserved every user across all three datasets.

1.3 The Item Cold-Start Problem

Once the split was built, I ran another sanity check that produced a genuinely useful number for the report:

How many distinct movies appear in the test set but were never rated in the training set at all?

train_movies = set(
    row.movieId
    for row in train.select("movieId").distinct().collect()
)

test_only_movies = (
    test
    .filter(~F.col("movieId").isin(train_movies))
    .select("movieId")
    .distinct()
    .count()
)

train.select("movieId").distinct() produces a Spark DataFrame containing every unique movieId in train. .collect() brings these IDs to the driver, where the generator expression extracts each movieId and set(...) converts them into a regular Python set called train_movies.

F.col("movieId").isin(train_movies) checks whether each test movie appears in the training set. The ~ negates this condition, so .filter(...) keeps only test rows whose movie never appeared in training.

.select("movieId").distinct().count() reduces these rows to unique movie IDs and counts them.

Result

901 movies out of 9,724 total movies (~9.3% of the catalog) appeared in the test set with zero training history.

This is therefore a real, measured instance of item cold start in our dataset — rather than simply a hypothetical problem.

ALS Cold-Start Strategy

This finding explains why we later use Spark ALS with:

coldStartStrategy="drop"

Those 901 movies have no learned latent vector because they never appeared in the training data.

ALS therefore cannot generate a meaningful prediction for those movie-user combinations, resulting in NaN predictions.

Using coldStartStrategy="drop removes these undefined predictions before calculating RMSE.This keeps the evaluation metric meaningful rather than allowing missing predictions to corrupt the calculation.

However, there is an important distinction between evaluation and production.

Dropping these movies is acceptable for evaluation because we are explicitly measuring model performance only where the model has enough information to make a prediction.

In a production recommender system, simply dropping a movie would not be an ideal solution. A genuinely new movie still needs to be recommendable.

Instead, these cold-start items would need to be assigned some initial information — such as basic popularity or rating-based features — so that the system can begin incorporating them into the recommendation ecosystem.

How this can be handled in a production setting is discussed later in Section X.

1.4 Concrete Data Split Example

Suppose Alice has 5 ratings, with timestamps in chronological order:

[100, 105, 200, 250, 300]

For example, she rated Titanic first, then Notebook, then Matrix, and so on.

1. Creating the Window

w = Window.partitionBy("userId").orderBy("timestamp")

This defines a window. Think of it as:

“Group the data by userId, then, within each group, sort by timestamp.”

The window does not compute anything by itself. Instead, it is a reusable sorting and grouping instruction that we can apply to functions below.

It is similar to saving an Excel rule such as:

“For each user, sort their ratings chronologically.”

2. Ranking Each User’s Ratings

ratings_ranked = (
    ratings
    .withColumn("rn", F.row_number().over(w))

F.row_number().over(w) assigns a sequential number to each row within each user’s group, following the timestamp order defined by w.

For Alice:

Her earliest rating gets rn = 1
Her second-earliest rating gets rn = 2
Her third-earliest rating gets rn = 3
Her fourth-earliest rating gets rn = 4
Her latest rating gets rn = 5

The important point is that this numbering happens independently for every user.

Because we used:

partitionBy("userId")

the numbering restarts at 1 for every user. We are not assigning one global sequence across the entire dataset.

3. Counting Each User’s Total Ratings

    .withColumn(
        "user_total",
        F.count("*").over(Window.partitionBy("userId"))
    )
)

This adds a user_total column showing how many ratings that user has in total.

For Alice, user_total = 5.

The value 5 appears on every one of Alice’s rows, because it is a property of her entire user group.

Unlike the previous window, this one does not use orderBy():

Window.partitionBy("userId")

We are not ranking or ordering anything here. We are simply counting the number of rows belonging to each user and attaching that count to every row for that user.

Result So Far for Alice

movieId	timestamp	rn	user_total
Titanic	100	1	5
Notebook	105	2	5
Matrix	200	3	5
Avengers	250	4	5
Star Wars	300	5	5

The Split-Point Calculation

Now that every rating has a sequential number and we know how many ratings each user has, we can determine where the train, validation, and test sets should begin.

4. Calculating the Test Cutoff

ratings_split = ratings_ranked.withColumn(
    "test_cutoff",
    F.ceil(F.col("user_total") * 0.10)
)

For Alice:

user_total = 5
test_cutoff = ceil(5 × 0.10)
             = ceil(0.5)
             = 1

Therefore, Alice’s most recent 1 rating is reserved for the test set.

The use of ceil() is important.

It guarantees that even users with a small number of ratings receive at least one rating in the test set.

Without ceil(), a calculation such as:

5 × 0.10 = 0.5

could be truncated to 0, leaving Alice with zero test ratings.

5. Calculating the Validation Cutoff

).withColumn(
    "val_cutoff",
    F.ceil(F.col("user_total") * 0.10)
)

We apply the same logic for the validation set.

For Alice:

val_cutoff = ceil(5 × 0.10)
           = ceil(0.5)
           = 1

So Alice will also have 1 rating reserved for validation.

6. Assigning the Train, Validation, and Test Labels

We now use the row number and the cutoff values to assign each rating to a split.

Test Set

.withColumn(
    "split",
    F.when(
        F.col("rn") > F.col("user_total") - F.col("test_cutoff"),
        "test"
    )

For Alice:

user_total - test_cutoff
= 5 - 1
= 4

The condition therefore becomes:

rn > 4

Only Alice’s row with rn = 5 satisfies this condition.

Therefore, her latest rating is assigned to the test set.

Validation Set

    .when(
        F.col("rn") >
        F.col("user_total") - F.col("test_cutoff") - F.col("val_cutoff"),
        "val"
    )

For Alice:

5 - 1 - 1 = 3

So the condition becomes:

rn > 3

However, the test condition is checked first.

Therefore:

rn = 5 → already assigned to test
rn = 4 → satisfies rn > 3, so assigned to val

Alice’s second-most-recent rating therefore becomes her validation rating.

Training Set

Finally:

    .otherwise("train")
)

Any rating that has not already been assigned to test or val is assigned to train.

For Alice, that means:

rn = 1 → train
rn = 2 → train
rn = 3 → train

These are her three earliest ratings.

Final Result for Alice

movieId	timestamp	rn	split
Titanic	100	1	train
Notebook	105	2	train
Matrix	200	3	train
Avengers	250	4	val
Star Wars	300	5	test

This gives us exactly the desired time-based split:

Earliest ratings → Train
Second-most-recent ratings → Validation
Most recent ratings → Test

The use of ceil() also ensures that users are not left with zero ratings in either validation or test, even when they have very few interactions.

In this example, Alice has only 5 ratings, so the 10% calculation would normally produce 0.5. Using ceil() turns that into 1, ensuring that she gets at least one rating in each held-out split.

Note: In the actual MovieLens dataset, the minimum number of ratings per user is 20, so the same logic applies comfortably to every user.

Why `partitionBy("userId")` Is So Important

This entire process runs independently for every user.

For example, Alice’s ratings are ranked:

1, 2, 3, 4, 5

But another user might have 20 ratings:

1, 2, 3, ..., 20

And another user might have 100 ratings:

1, 2, 3, ..., 100

The numbering, counting, and split calculation are all performed within each user’s own group.

This is exactly what Window.partitionBy("userId") ensures. We never mix Alice’s ratings with another user’s ratings. Each user’s interaction history is treated independently, allowing us to create a personalized chronological train/validation/test split for all 610 users simultaneously.

1. Data Preprocessing

1.1. Preliminary Inspection

ML Small Dataset from Group Lens: 100,000 ratings and 3,600 tag applications applied to 9,000 movies by 600 users. Last updated 9/2018.

User Ids

Movie Ids

Ratings Data File Structure (ratings.csv)

userId,movieId,rating,timestamp

Ratings Table

Tags Data File Structure (tags.csv)

userId,movieId,tag,timestamp

Movies Data File Structure (movies.csv)

movieId,title,genres

Movies Table

Links Data File Structure (links.csv)

movieId,imdbId,tmdbId

Preliminary Observation Summary

File formats - csv vs parquet

1.2. Data Splitting Strategy

Potential problems with random split

Problem I - Blindspots

Problem II- Temporal data

Movie-Lens split solution

The fix I used : Per-user, chronological splitting.

Key Spark Operations Used Here

Window.partitionBy("userId").orderBy("timestamp")

F.row_number().over(w)

F.count("*").over(Window.partitionBy("userId"))

F.ceil(...)

F.when(...).otherwise(...)

GROUP BY vs. PARTITION BY

Split Result

1.3 The Item Cold-Start Problem

Result

ALS Cold-Start Strategy

1.4 Concrete Data Split Example

1. Creating the Window

2. Ranking Each User’s Ratings

3. Counting Each User’s Total Ratings

Result So Far for Alice

The Split-Point Calculation

4. Calculating the Test Cutoff

5. Calculating the Validation Cutoff

6. Assigning the Train, Validation, and Test Labels

Test Set

Validation Set

Training Set

Final Result for Alice

Why partitionBy("userId") Is So Important

`Window.partitionBy("userId").orderBy("timestamp")`

`F.row_number().over(w)`

`F.count("*").over(Window.partitionBy("userId"))`

`F.ceil(...)`

`F.when(...).otherwise(...)`

`GROUP BY` vs. `PARTITION BY`

Why `partitionBy("userId")` Is So Important