Spark ML Recommendation, in depth: ALS collaborative filtering from first principles to production top-K serving

The problem: filling in a sparse matrix

Picture a matrix R with one row per user and one column per item. A cell holds a rating, or a count of plays, or nothing at all. Real matrices are more than 99 percent empty: a music service with two million users and two hundred thousand tracks has 400 billion cells and perhaps 300 million observed ones. Recommending means filling in the blanks.

Matrix factorization assumes the matrix is approximately low rank. Each user u gets a vector x_u of k numbers and each item i gets a vector y_i of the same length, and the predicted preference is their dot product. The k dimensions are latent factors. With k = 64 the model stores 64 numbers per user and per item instead of 400 billion cells, and that compression is what lets it generalise to cells it never saw.

Training minimises squared error on the observed cells plus a penalty on the size of the vectors:

minimise  sum over observed (u,i) of (r_ui - x_u . y_i)^2
        + lambda * ( sum_u n_u * |x_u|^2  +  sum_i n_i * |y_i|^2 )

n_u = number of ratings by user u,  n_i = number of ratings of item i

The n_u and n_i weights are the "weighted lambda" scheme (ALS-WR) that Spark uses: the documentation states that regParam is scaled by the number of ratings a user generated or an item received. The practical consequence is that a regParam tuned on a sample transfers reasonably well to the full data set.

How ALS trains: alternating closed-form solves

The objective is not convex in x and y together, which is why plain least squares cannot solve it in one step. But if you freeze every item vector, each user's vector becomes an ordinary ridge regression with a closed-form answer, and every user can be solved independently. Freeze the users and the same is true for items. ALS alternates between the two:

initialise item factors Y randomly (seeded)
repeat maxIter times:
    for each user u in parallel:
        Y_u = rows of Y for items u interacted with
        x_u = solve( Y_u^T Y_u + lambda * n_u * I ,  Y_u^T r_u )     # k x k system
    for each item i in parallel:
        X_i = rows of X for users who interacted with i
        y_i = solve( X_i^T X_i + lambda * n_i * I ,  X_i^T r_i )

Each step solves a k-by-k linear system, so the cost per user is roughly the number of that user's interactions times k squared, plus k cubed for the solve. There is no learning rate and no gradient. With nonnegative=True Spark swaps the plain solve for a non-negative least squares solver, which gives more interpretable factors at some cost in accuracy and speed.

Implicit feedback changes the objective. Following Hu, Koren and Volinsky's paper, which the Spark docs cite, every cell becomes a binary preference p_ui (1 if the user interacted at all, otherwise 0) with a confidence c_ui = 1 + alpha × r_ui. All cells now count, including the empty ones, which carry preference 0 at confidence 1. That sounds impossible at 400 billion cells, but the algebra trick is that Y^TC_uY equals Y^TY plus a correction that touches only the user's observed items, and Y^TY is computed once per iteration. The cost stays proportional to the observed data.

Distribution. Spark partitions users into numUserBlocks blocks and items into numItemBlocks blocks (both default 10). Before the first iteration it builds routing tables recording which item blocks each user block needs. Every half-iteration then sends each item factor only to the user blocks that reference it, solves locally and reverses direction. Shuffle volume per half-iteration is therefore about the number of distinct (block, item) pairs times k times 4 bytes, not the number of ratings.

Architecture: where ALS sits in the pipeline

The training job is the middle of a longer pipeline. Upstream, raw events are deduplicated and collapsed into one row per (user, item) with a strength value. A mapping table translates string identifiers into dense integers, because the DataFrame API only supports integer user and item IDs (other numeric types are accepted if their values fit in the 32-bit integer range, and the indices must not be negative). Downstream, precomputed lists are filtered against items the user already consumed and written to a store the serving tier reads in a single lookup.

A complete implicit-feedback job

Here is a complete implicit-feedback job. It builds stable integer indices, aggregates plays, splits by time rather than at random, trains, and evaluates with a ranking metric.

from pyspark.sql import SparkSession, functions as F, Window
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RankingEvaluator

spark = SparkSession.builder.appName("als-train").getOrCreate()
spark.sparkContext.setCheckpointDir("s3://bucket/checkpoints/als")   # needed for checkpointInterval

events = spark.read.parquet("s3://bucket/events/plays")   # user_id, track_id, ts, ms_played

# 1. Dense integer ids. Persist this table: indices must stay stable between retrains.
def index(col):
    w = Window.orderBy(col)
    return events.select(col).distinct().withColumn(col + "_idx", F.row_number().over(w) - 1)

users, items = index("user_id"), index("track_id")

# 2. One row per (user, item); a play counts only if it lasted at least 30 seconds.
cutoff = "2026-09-20"
inter = (events.filter(F.col("ms_played") >= 30000)
         .join(users, "user_id").join(items, "track_id")
         .groupBy("user_id_idx", "track_id_idx")
         .agg(F.count("*").alias("plays"), F.max("ts").alias("last_ts"))
         .withColumn("strength", F.log1p("plays")))         # damp heavy repeaters

train = inter.filter(F.col("last_ts") < cutoff)
test = inter.filter(F.col("last_ts") >= cutoff)

als = ALS(userCol="user_id_idx", itemCol="track_id_idx", ratingCol="strength",
          implicitPrefs=True, rank=64, regParam=0.05, alpha=20.0,
          maxIter=15, checkpointInterval=5, coldStartStrategy="drop", seed=7)
model = als.fit(train)

# 3. Ranking evaluation: top-50 predicted items vs items actually played after the cutoff.
recs = model.recommendForUserSubset(test.select("user_id_idx").distinct(), 50)
pred = recs.select("user_id_idx",
                   F.expr("transform(recommendations, r -> cast(r.track_id_idx as double))").alias("prediction"))
truth = test.groupBy("user_id_idx").agg(
    F.collect_list(F.col("track_id_idx").cast("double")).alias("label"))
ndcg = RankingEvaluator(metricName="ndcgAtK", k=50).evaluate(pred.join(truth, "user_id_idx"))
print("NDCG@50", ndcg)

model.write().overwrite().save("s3://bucket/models/als/2026-10-04")

Four details matter. setCheckpointDir is not optional for long runs: without it Spark ignores checkpointInterval, the lineage grows every iteration, and deep jobs can end in a StackOverflowError. The log1p transform stops one user who looped a track 4,000 times from dominating the confidence weights. RankingEvaluator (since Spark 3.0, still marked experimental) expects array columns of doubles for both prediction and label, hence the casts. A production version should also drop from the predictions any track the user played before the cutoff, otherwise it scores itself on re-recommending history.

Worked example: sizing a music catalogue

Size the music example before you request a cluster. Two million users and 200,000 tracks at rank 64 give 2.2 million vectors of 64 single-precision floats, about 563 MB of factors in total. That is small. The expensive parts are elsewhere.

Training. With 300 million interactions, one user half-iteration does about 300 million × 64² ≈ 1.2 × 10¹² multiply-adds to build the normal equations, plus two million 64-by-64 solves. If iterations take far longer, look at shuffle spill and skew before you add hardware.

Batch scoring. recommendForAllUsers(50) has to score every user against every item: 2 million × 200,000 = 4 × 10¹¹ dot products of length 64, roughly 2.6 × 10¹³ multiply-adds. Spark stacks vectors into blocks (the blockSize parameter, default 4096) so the work runs as matrix multiplies and keeps a bounded top-K heap per user. It is still the stage most likely to dominate runtime, and it is the one to move to an approximate nearest-neighbour index if the catalogue grows into the millions.

Confidence arithmetic. With alpha = 20, a user with 30 plays gets strength log1p(30) ≈ 3.43 and confidence 1 + 20 × 3.43 ≈ 70; a single play gives about 15; every unobserved track sits at confidence 1. The model works hardest to fit heavy listening and treats absence as weak negative evidence, which is exactly the intent.

Tuning the parameters that matter

Tune on a ranking metric with a time split, not on RMSE. CrossValidator and TrainValidationSplit work with ALS, but their random folds leak future interactions into training, so many teams write a short loop over a parameter grid against one fixed temporal holdout instead. Search alpha and regParam together; they trade off, and the best rank usually depends on both. If you wrap ALS in a Spark ML Pipeline, the ID indexing can live in the same fitted pipeline as the model.

Failure modes

• NaN predictions. With the default coldStartStrategy="nan", users or items absent from training get NaN, and one NaN turns an RMSE into NaN. Use "drop" for evaluation; in serving, treat NaN as the signal to fall back.
• Unstable indices. Re-deriving indices with row_number over a changed set of users shifts every index. Factors saved last week then belong to different people. Persist the mapping and only append to it.
• IDs outside the integer range. Casting 64-bit hashes or large account numbers to int truncates them and silently merges users. Map to dense indices instead.
• Explicit settings on implicit data. Fitting play counts with implicitPrefs=False treats 40 plays as a rating of 40 and ignores the zeros. The model trains, the loss looks fine and recommendations are poor.
• Evaluation leakage. A random split puts a user's next week in training. Offline metrics jump; online results do not follow.
• Bots and power users. A handful of accounts with millions of events skews item factors and creates one huge partition. Cap per-user events or the strength value, and filter obvious automation.
• Popularity collapse. Low regParam and high alpha tend to push everyone toward the same hits. Track catalogue coverage alongside NDCG.
• Lineage overflow and executor loss. Long runs without a checkpoint directory, or with intermediate storage set to memory only, fail late and expensively. Keep the default MEMORY_AND_DISK storage levels unless you have measured a reason to change them.

Serving and operations

Most deployments precompute. A nightly or hourly job retrains, scores active users with recommendForUserSubset, removes already-consumed items, keeps the top 100 to 500 per user and bulk-loads a key-value store. The serving API reads one key, applies business filters such as availability and region, and reranks if a downstream model exists. New users and users the model has not seen get a popularity or content-based fallback until they appear in a retrain.

Version everything that must agree: the model directory, the ID mapping snapshot it was trained with, and the parameter set. A saved ALSModel is a directory of user and item factor tables, so you can also export itemFactors to a vector index and compute recommendations on request from a user vector. That is how you serve users whose vector you refresh more often than the whole model. Watch training-time metrics (per-iteration duration, shuffle bytes, spill) and online metrics (click-through on recommendation slots, coverage, share of requests served by fallback). A sudden jump in fallback share usually means the mapping and the model drifted apart.

Trade-offs and related reading

ALS is the right first model when you have interaction logs, want something explainable that trains on existing Spark infrastructure, and can tolerate batch freshness. Its limits are structural: it ignores context such as time of day and device, it cannot use item content, so a new item is invisible until people interact with it, and its scores are relative affinities rather than probabilities. Two-tower neural retrieval models and sequence models address those gaps at much higher engineering cost.

For the surrounding system, read how a large service layers candidate generation, ranking and filtering in the Netflix recommendation architecture, compare approaches in the general recommender systems overview, and see where ALS sits among Spark's other algorithms in the MLlib guide.

What to do next

1. Decide whether your data is explicit or implicit, and set implicitPrefs to match before anything else.
2. Build and persist a dense integer ID mapping; make retrains append to it rather than rebuild it.
3. Aggregate events to one row per user and item, and damp extreme counts with a log or a cap.
4. Create a temporal holdout and evaluate with RankingEvaluator (NDCG or precision at the K you serve), excluding items already consumed.
5. Set a checkpoint directory and a checkpointInterval before running more than about ten iterations.
6. Grid-search rank, regParam and alpha against the holdout; record coverage as well as accuracy.
7. Precompute top-K lists, filter seen items, load a key-value store and wire in a popularity fallback for cold-start users.
8. Version the model with its mapping snapshot, and alert on fallback share and scoring-stage runtime.