MovieLens Example

In this notebook, we train an AverageModel on the MovieLens dataset with a BPRLoss.

The pipeline is made of 4 steps

  • step 1: given the MovieLens ratings.csv file, create tfrecords for the training, evaluation and test sets.

  • step 2: train an AverageModel (optionally, use tf-yarn to distribute training and evaluation on a cluster) and export the embeddings as well as the graph.

  • step 3: write predictions, i.e. for all the test timelines, compute the user representation.

  • step 4: evaluate predictions, i.e. look how similar the recommendations provided by the model are to the actual movies seen by the user in the “future”

Install

[ ]:
!pip install deepr[cpu] faiss_cpu

Download the dataset

First download the movielens dataset

[ ]:
!wget http://files.grouplens.org/datasets/movielens/ml-20m.zip
!unzip ml-20m.zip
[ ]:
import os
import logging

import tensorflow as tf

import deepr
from deepr.examples import movielens
[3]:
logging.basicConfig(level=logging.INFO)
[4]:
dataset_path = os.getcwd() + "/ml-20m"
path_ratings = f"{dataset_path}/ratings.csv"
path_root = "average"
path_model = path_root + "/model"
path_data = path_root + "/data"
path_variables = path_root + "/variables"
path_predictions = path_root + "/predictions.parquet.snappy"
path_saved_model = path_root + "/saved_model"
path_mapping = path_data + "/mapping.txt"
path_train = path_data + "/train.tfrecord.gz"
path_eval = path_data + "/eval.tfrecord.gz"
path_test = path_data + "/test.tfrecord.gz"
deepr.io.Path(path_root).mkdir(exist_ok=True)
deepr.io.Path(path_model).mkdir(exist_ok=True)
deepr.io.Path(path_data).mkdir(exist_ok=True)
max_steps = 100_000

Step 1 - Build TF Records

The job takes as input the csv ratings and create 3 tfrecords files (train, validation, test).

Each file contains timeline of user ratings split into input and target.

[5]:
build = movielens.jobs.BuildRecords(
    path_ratings=path_ratings,
    path_mapping=path_mapping,
    path_train=path_train,
    path_eval=path_eval,
    path_test=path_test,
    min_rating=4,
    min_length=5,
    num_negatives=8,
    target_ratio=0.2,
    size_test=10_000,
    size_eval=10_000,
    shuffle_timelines=True,
    seed=2020,
)

Now that the job is defined, let’s run it.

We run it before defining the other jobs because we need to know the vocabulary size (number of movies in the dataset) to build the embedding matrix.

[ ]:
build.run()

Step 2 - Training

This defines a training job to build a model that transform a timeline of movies embedding to an user embedding. It takes as input the tf records and upon completion writes 3 artifacts : * dataframe of the biases and the embeddings * saved model : protobuf containing the model definition and weights

In this specific instance we train an average model with a BPR loss and compute a triple precision on the validation set.

[9]:
train = deepr.jobs.Trainer(
    path_model=path_model,
    pred_fn=movielens.layers.AverageModel(vocab_size=deepr.vocab.size(path_mapping), dim=100, keep_prob=0.5),
    loss_fn=movielens.layers.Loss(loss="bpr", vocab_size=deepr.vocab.size(path_mapping)),
    optimizer_fn=deepr.optimizers.TensorflowOptimizer("LazyAdam", 0.001),
    train_input_fn=deepr.readers.TFRecordReader(path_train),
    eval_input_fn=deepr.readers.TFRecordReader(path_eval, shuffle=False),
    prepro_fn=movielens.prepros.RecordPrepro(
        min_input_size=3,
        min_target_size=3,
        max_input_size=50,
        max_target_size=50,
        buffer_size=1024,
        batch_size=128,
        repeat_size=None,
        prefetch_size=1,
        num_parallel_calls=8,
    ),
    train_spec=deepr.jobs.TrainSpec(max_steps=max_steps),
    eval_spec=deepr.jobs.EvalSpec(steps=None, start_delay_secs=30, throttle_secs=30),
    final_spec=deepr.jobs.FinalSpec(steps=None),
    exporters=[
        # The training will keep the model with the best triplet precision
        deepr.exporters.BestCheckpoint(metric="triplet_precision", mode="increase"),
        # Export biases and embeddings as a dataframe
        deepr.exporters.SaveVariables(path_variables=path_variables, variable_names=["biases", "embeddings"]),
        # Export a saved model using specified fields as input
        deepr.exporters.SavedModel(
            path_saved_model=path_saved_model,
            fields=[
                deepr.Field(name="inputPositives", shape=(None,), dtype=tf.int64),
                deepr.Field(name="inputMask", shape=(None,), dtype=tf.bool),
            ],
        ),
    ],
    train_hooks=[
        # Log metrics, hyperparams, initial values to the console, and optionally mlflow and graphite
        deepr.hooks.LoggingTensorHookFactory(
            name="training",
            functions={
                "memory_gb": deepr.hooks.ResidentMemory(unit="gb"),
                "max_memory_gb": deepr.hooks.MaxResidentMemory(unit="gb"),
            },
            every_n_iter=300,
            use_graphite=False,
            use_mlflow=False,
        ),
        deepr.hooks.SummarySaverHookFactory(save_steps=300),
        deepr.hooks.NumParamsHook(use_mlflow=False),
        deepr.hooks.LogVariablesInitHook(use_mlflow=False),
        deepr.hooks.StepsPerSecHook(
            name="training",
            batch_size=128,
            every_n_steps=300,
            skip_after_step=max_steps,
            use_mlflow=False,
            use_graphite=False,
        ),
        # Stop the training if triplet precision does not improve
        deepr.hooks.EarlyStoppingHookFactory(
            metric="triplet_precision",
            mode="increase",
            max_steps_without_improvement=1000,
            min_steps=5_000,
            run_every_steps=300,
            final_step=max_steps,
        ),
    ],
    eval_hooks=[deepr.hooks.LoggingTensorHookFactory(name="validation", at_end=True)],
    final_hooks=[deepr.hooks.LoggingTensorHookFactory(name="final_validation", at_end=True)],
    train_metrics=[deepr.metrics.StepCounter(name="num_steps"), deepr.metrics.DecayMean(tensors=["loss"], decay=0.98)],
    eval_metrics=[deepr.metrics.Mean(tensors=["loss", "triplet_precision"])],
    final_metrics=[deepr.metrics.Mean(tensors=["loss", "triplet_precision"])],
    run_config=deepr.jobs.RunConfig(
        save_checkpoints_steps=300, save_summary_steps=300, keep_checkpoint_max=None, log_step_count_steps=300
    ),
    config_proto=deepr.jobs.ConfigProto(
        inter_op_parallelism_threads=8, intra_op_parallelism_threads=8, gpu_device_count=0, cpu_device_count=48,
    ),
)

Step 3 - Predict job

The predict job reloads the test dataset and the saved model, performs the inference and writes the user embeddings in a dataframe to be reused later by the validation job.

[10]:
predict = movielens.jobs.Predict(
    path_saved_model=path_saved_model,
    path_predictions=path_predictions,
    input_fn=deepr.readers.TFRecordReader(path_test, shuffle=False),
    prepro_fn=movielens.prepros.RecordPrepro(),
)

Step 4 - Evaluation job

This job takes user embedding, and uses faiss to retrieve the k nearest neighboors in the product embedding space, and compute metrics using the target timelines.

[11]:
evaluate = [
    movielens.jobs.Evaluate(
        path_predictions=path_predictions,
        path_embeddings=path_variables + "/embeddings",
        path_biases=path_variables + "/biases",
        k=k,
    )
    for k in [10, 20, 50]
]

Run the pipeline

All the jobs definition are lazy, and so is the pipeline. Calling run on it will actually perform all these steps.

[12]:
pipeline = deepr.jobs.Pipeline([train, predict] + evaluate)
[ ]:
pipeline.run()

You should expecte the following metrics

INFO:deepr.examples.movielens.jobs.evaluate:precision@10 = 0.15774000000000002
recall@10 = 0.22966825396825397
f1@10 = 0.17782589040302665
NDCG@10 = 0.218460915855854
INFO:deepr.examples.movielens.jobs.evaluate:precision@20 = 0.12564500000000003
recall@20 = 0.29865131767226577
f1@20 = 0.16083200785947185
NDCG@20 = 0.24568841120719748
INFO:deepr.examples.movielens.jobs.evaluate:precision@50 = 0.08614800000000002
recall@50 = 0.44487653142758526
f1@50 = 0.1303801216635994
NDCG@50 = 0.3022301157999229

You can find configs on github that achieve better performance.

Visualize embeddings with a KNN

Let’s check if the movie embeddings produce make sense * load the movie embeddings * load the movie title * build a knn index on that * do a query with a known movie to check that its closest neighboors make sense

[14]:
import numpy as np
import pyarrow.parquet as pq
import faiss
import pyarrow.csv as pc
from IPython.display import display
import pandas as pd
[17]:
embeddings = np.vstack(pq.read_table("average/variables/embeddings").to_pandas().to_numpy())
mapp = {int(movie_id):indice for indice, movie_id in enumerate(open("average/data/mapping.txt", "r").read().split("\n"))}
inversed_map = {indice:movie_id for movie_id, indice in mapp.items()}
index = faiss.IndexFlatIP(embeddings.shape[-1])
index.add(np.ascontiguousarray(embeddings))

def knn_query(index, query, ksearch):
    D, I = index.search(np.expand_dims(query,0), ksearch)
    distances = D[0]
    product_indices = I[0]
    product_ids = [inversed_map[i] for i in product_indices]
    return list(zip(product_ids, distances))

movies = pc.read_csv(f"{dataset_path}/movies.csv").to_pandas()

def display_results_df(results):
    data = [[movies[movies.movieId == movie_id]["genres"].to_numpy()[0], movies[movies.movieId == movie_id]["title"].to_numpy()[0], distance] for movie_id, distance in results]

    df = pd.DataFrame(data, columns = ['Genre', 'Title', 'Distance'])
    display(df)

def display_movie(movie_id):
    data = [[(movies[movies.movieId == movie_id]["genres"].to_numpy()[0]), (movies[movies.movieId == movie_id]["title"].to_numpy()[0])]]

    df = pd.DataFrame(data, columns = ['Genre', 'Title'])
    display(df)
[18]:
p = movies[movies.title.str.lower().str.contains("star wars")]["movieId"].to_numpy()[0]
print("Query")
display_movie(p)
print("Knn results")
display_results_df(knn_query(index, embeddings[mapp[p]], 5))
Query
Genre Title
0 Action|Adventure|Sci-Fi Star Wars: Episode IV - A New Hope (1977)
Knn results
Genre Title Distance
0 Action|Adventure|Sci-Fi Star Wars: Episode IV - A New Hope (1977) 16.765795
1 Action|Adventure|Sci-Fi Star Wars: Episode VI - Return of the Jedi (1983) 14.477997
2 Action|Adventure|Sci-Fi Star Wars: Episode V - The Empire Strikes Back... 13.837748
3 Action|Adventure|Sci-Fi|Thriller Star Trek: First Contact (1996) 13.092430
4 Children|Comedy|Fantasy|Musical Willy Wonka & the Chocolate Factory (1971) 11.858896

As you can see, the genre of the nearest neighboors is similar and the movies are related to the movie query