{ "cells": [ { "cell_type": "markdown", "id": "82944a1d", "metadata": {}, "source": [ "\n", "" ] }, { "cell_type": "markdown", "id": "08dc7f2b", "metadata": {}, "source": [ "# Travaux pratiques - Recommandation avec Spark MLlib et GraphFrames\n", "\n", "Références externes utiles :\n", "\n", "- [Documentation Spark](https://spark.apache.org/docs/latest/) \n", "- [Démarrage avec les DataFrames Spark en Python](https://spark.apache.org/docs/latest/api/python/getting_started/quickstart_df.html) \n", "- [Documentation API Spark en Python](https://spark.apache.org/docs/latest/api/python/index.html) \n", "- [Tutoriel Spark avec exemples](https://sparkbyexamples.com/pyspark-tutorial/) \n", "- [Site langage Python](https://www.python.org/) " ] }, { "cell_type": "markdown", "id": "034d3b06", "metadata": {}, "source": [ "## Objectif du TP\n", "\n", "Ce TP met en œuvre un système de recommandation de films complet sur le jeu de\n", "données **MovieLens** avec Spark MLlib. Le travail est décomposé en quatre\n", "parties :\n", "\n", "1. Exploration des données et structure de la matrice d’utilités. \n", "1. Factorisation matricielle avec ALS : entraînement, évaluation, recommandations. \n", "1. Réglage des hyperparamètres : impact du rang et de la régularisation. \n", "1. Recommandation par graphe : PPR et filtrage collaboratif via GraphFrames. \n", "\n", "\n", "Les données MovieLens ont été vues dans le\n", "TP de préparation Python. Nous tentons d’utiliser ici\n", "la version complète **MovieLens 25M** (25 millions de notes, 162 000 utilisateurs,\n", "62 000 films) pour que la parallélisation de Spark soit significative." ] }, { "cell_type": "markdown", "id": "234aceb4", "metadata": {}, "source": [ "## Partie 1 : Données et exploration" ] }, { "cell_type": "markdown", "id": "9a98d874", "metadata": {}, "source": [ "### Vérification des bibliothèques\n", "\n", "La communication entre Spark et Polars exige la version 2 de `sparkpl` et la version 4.x de Pyspark, une vérification est nécessaire :" ] }, { "cell_type": "code", "execution_count": null, "id": "40516a28", "metadata": { "hide-output": false }, "outputs": [], "source": [ "import sys\n", "!{sys.executable} -m pip install \"pyspark>=4,<5\" sparkpl==2.0.1 polars altair graphframes-py==0.12.1" ] }, { "cell_type": "markdown", "id": "1bede4a0", "metadata": {}, "source": [ "Si la réponse vous indique que `pyspark` était déjà installé alors vous pouvez poursuivre. En revanche, si cette instruction vient d’installer la version 4 à la place d’une plus ancienne alors il faut **redémarrer le noyau (icône circulaire du menu en haut de la page) avant de poursuivre !**" ] }, { "cell_type": "markdown", "id": "e71f92c1", "metadata": {}, "source": [ "### Récupération des données MovieLens 25M" ] }, { "cell_type": "code", "execution_count": null, "id": "32341c0c", "metadata": { "hide-output": false }, "outputs": [], "source": [ "import os, sys\n", "os.environ['PYSPARK_PYTHON'] = sys.executable\n", "os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable" ] }, { "cell_type": "code", "execution_count": null, "id": "7ac40429", "metadata": { "hide-output": false }, "outputs": [], "source": [ "dossier = \"tpRecommandation/movielens/\"\n", "os.makedirs(dossier, exist_ok=True)\n", "\n", "archive = dossier + \"ml-25m.zip\"\n", "if not os.path.exists(dossier + \"ml-25m/ratings.csv\"):\n", " if not os.path.exists(archive):\n", " print(\"Téléchargement MovieLens 25M (~250 Mo)...\")\n", " os.system(f\"wget -q https://files.grouplens.org/datasets/movielens/ml-25m.zip -P {dossier}\")\n", " print(\"Décompression...\")\n", " os.system(f\"unzip -q {archive} -d {dossier}\")\n", " print(\"OK\")\n", "else:\n", " print(\"Données déjà présentes.\")" ] }, { "cell_type": "markdown", "id": "1636dadd", "metadata": {}, "source": [ "### Initialisation de Spark" ] }, { "cell_type": "code", "execution_count": null, "id": "a6f6735a", "metadata": { "hide-output": false }, "outputs": [], "source": [ "from importlib.metadata import version\n", "from pyspark.sql import SparkSession\n", "from pyspark.sql import functions as F\n", "\n", "gf_version = version(\"graphframes-py\") # garde le JAR aligné sur le paquet pip\n", "\n", "spark = (SparkSession.builder\n", " .appName(\"tp-recommandation\")\n", " .config(\"spark.driver.memory\", \"4g\")\n", " .config(\"spark.jars.packages\",\n", " f\"io.graphframes:graphframes-spark4_2.13:{gf_version}\")\n", " .getOrCreate())" ] }, { "cell_type": "markdown", "id": "2fe202f3", "metadata": {}, "source": [ "### Chargement des données" ] }, { "cell_type": "code", "execution_count": null, "id": "6b42b14e", "metadata": { "hide-output": false }, "outputs": [], "source": [ "ratings = spark.read.csv(\n", " dossier + \"ml-25m/ratings.csv\",\n", " header=True, inferSchema=True)\n", "movies = spark.read.csv(\n", " dossier + \"ml-25m/movies.csv\",\n", " header=True, inferSchema=True)\n", "\n", "print(\"Schéma ratings :\"), ratings.printSchema()\n", "print(\"Schéma movies :\"), movies.printSchema()" ] }, { "cell_type": "code", "execution_count": null, "id": "07b1216e", "metadata": { "hide-output": false }, "outputs": [], "source": [ "n_ratings = ratings.count()\n", "n_users = ratings.select(\"userId\").distinct().count()\n", "n_movies = ratings.select(\"movieId\").distinct().count()\n", "\n", "print(f\"Notes : {n_ratings:,}\")\n", "print(f\"Utilisateurs: {n_users:,}\")\n", "print(f\"Films : {n_movies:,}\")\n", "print(f\"Taux de remplissage : {n_ratings / (n_users * n_movies):.6f}\")" ] }, { "cell_type": "markdown", "id": "c85af7a2", "metadata": {}, "source": [ "### Question\n", "\n", "Calculez le nombre de notes par film et ensuite par utilisateur. Calculez la médiane,\n", "la moyenne et la valeur maximale du nombre de notes par utilisateur. Comment\n", "caractériser la distribution ?" ] }, { "cell_type": "markdown", "id": "791c494a", "metadata": {}, "source": [ "### Distribution des notes" ] }, { "cell_type": "code", "execution_count": null, "id": "fa33e756", "metadata": { "hide-output": false }, "outputs": [], "source": [ "import polars as pl\n", "from sparkpl.converter import spark_to_polars\n", "import altair as alt\n", "alt.data_transformers.enable(\"default\", max_rows=None)\n", "\n", "# Distribution des notes (agrégation Spark, visualisation Altair)\n", "distrib = (ratings\n", " .groupBy(\"rating\")\n", " .agg(F.count(\"*\").alias(\"nb\"))\n", " .orderBy(\"rating\"))\n", "\n", "distrib_pl = spark_to_polars(distrib)\n", "\n", "alt.Chart(distrib_pl).mark_bar().encode(\n", " x=alt.X(\"rating:O\", title=\"Note\"),\n", " y=alt.Y(\"nb:Q\", title=\"Nombre de notes\"),\n", " tooltip=[\"rating:O\", alt.Tooltip(\"nb:Q\", format=\",\")]\n", ").properties(title=\"Distribution des notes pour MovieLens 25M\", width=400, height=280)" ] }, { "cell_type": "markdown", "id": "42d9f1f7", "metadata": {}, "source": [ "### Question\n", "\n", "Observez-vous un biais vers les notes élevées (*positivity bias*) ? Comment\n", "cela peut affecter un système de recommandation ?" ] }, { "cell_type": "markdown", "id": "64a07e9e", "metadata": {}, "source": [ "### Films les plus notés et les mieux notés" ] }, { "cell_type": "code", "execution_count": null, "id": "e4fc4c07", "metadata": { "hide-output": false }, "outputs": [], "source": [ "stats_films = (ratings\n", " .groupBy(\"movieId\")\n", " .agg(F.count(\"rating\").alias(\"nb_notes\"),\n", " F.mean(\"rating\").alias(\"note_moy\"))\n", " .join(movies.select(\"movieId\", \"title\", \"genres\"), on=\"movieId\")\n", ")\n", "\n", "# Top 10 films par nombre de notes\n", "print(\"Films les plus notés :\")\n", "stats_films.orderBy(F.desc(\"nb_notes\")).show(10, truncate=False)\n", "\n", "# Top 10 films les mieux notés (parmi ceux ayant au moins 1000 notes)\n", "print(\"Films les mieux notés (≥ 1000 notes) :\")\n", "(stats_films\n", " .filter(F.col(\"nb_notes\") >= 1000)\n", " .orderBy(F.desc(\"note_moy\"))\n", " .show(10, truncate=False))" ] }, { "cell_type": "markdown", "id": "8941dc4a", "metadata": {}, "source": [ "### Question\n", "\n", "Pourquoi c’est important de filtrer les films ayant peu de notes avant de\n", "calculer un classement par note moyenne ? Que se passerait-il sans ce filtre ?" ] }, { "cell_type": "markdown", "id": "658a3235", "metadata": {}, "source": [ "## Partie 2 : Factorisation matricielle avec ALS" ] }, { "cell_type": "markdown", "id": "c601fb72", "metadata": {}, "source": [ "### Rappel du principe\n", "\n", "ALS cherche des vecteurs latents $ \\mathbf{u}_i $ (utilisateurs) et\n", "$ \\mathbf{a}_j $ (films) de dimension $ m $ minimisant :\n", "\n", "$$\n", "\\sum_{\\text{Présent}(i,j)} (x_{ij} - \\mathbf{u}_i^T \\cdot \\mathbf{a}_j)^2\n", "+ \\lambda\\!\\left(\\sum_i n_i\\|\\mathbf{u}_i\\|^2 + \\sum_j n_j\\|\\mathbf{a}_j\\|^2\\right)\n", "$$\n", "\n", "La prédiction de note pour un couple [utilisateur, film] non observé est\n", "$ \\hat{x}_{kl} = \\mathbf{u}_k^T \\cdot \\mathbf{a}_l $." ] }, { "cell_type": "markdown", "id": "cda02dbc", "metadata": {}, "source": [ "### Séparation en apprentissage et test" ] }, { "cell_type": "code", "execution_count": null, "id": "258f977b", "metadata": { "hide-output": false }, "outputs": [], "source": [ "# Partitionnement 80/20\n", "train, test = ratings.randomSplit([0.8, 0.2], seed=42)\n", "train.cache()\n", "print(f\"Train : {train.count():,} | Test : {test.count():,}\")" ] }, { "cell_type": "markdown", "id": "caacf1f0", "metadata": {}, "source": [ "### Entraînement du modèle\n", "\n", "Pour un premier essai nous choisissons *a priori* les valeurs des paramètres :" ] }, { "cell_type": "code", "execution_count": null, "id": "f44066ca", "metadata": { "hide-output": false }, "outputs": [], "source": [ "from pyspark.ml.recommendation import ALS\n", "from pyspark.ml.evaluation import RegressionEvaluator\n", "\n", "als = ALS(\n", " rank=20, # dimension de l'espace latent\n", " maxIter=10, # nombre d'itérations ALS\n", " regParam=0.1, # paramètre de régularisation lambda\n", " userCol=\"userId\",\n", " itemCol=\"movieId\",\n", " ratingCol=\"rating\",\n", " coldStartStrategy=\"drop\" # ignorer les utilisateurs et films hors du train\n", ")\n", "\n", "model = als.fit(train)\n", "print(\"Entraînement terminé.\")" ] }, { "cell_type": "markdown", "id": "56cc8d52", "metadata": {}, "source": [ ">**Note**\n", ">\n", ">L’option `coldStartStrategy=\"drop\"` supprime du jeu de test les lignes\n", "correspondant à des utilisateurs et à des films non vus à l’entraînement\n", "(problème de démarrage à froid). L’alternative `\"nan\"` conserve ces lignes\n", "avec une prédiction NaN, ce qui est utile pour les analyser ultérieurement." ] }, { "cell_type": "markdown", "id": "16641e6b", "metadata": {}, "source": [ "### Évaluation sur le jeu de test\n", "\n", "Evaluation d’un modèle naïf (*baseline*) qui prédit toujours la note moyenne globale :" ] }, { "cell_type": "code", "execution_count": null, "id": "ecd8134c", "metadata": { "hide-output": false }, "outputs": [], "source": [ "evaluator_rmse = RegressionEvaluator(\n", " metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n", "evaluator_mae = RegressionEvaluator(\n", " metricName=\"mae\", labelCol=\"rating\", predictionCol=\"prediction\")\n", "\n", "mu = ratings.select(F.mean(\"rating\")).collect()[0][0]\n", "naive_preds = test.withColumn(\"prediction\", F.lit(mu))\n", "rmse_naive = evaluator_rmse.evaluate(naive_preds)\n", "mae_naive = evaluator_mae.evaluate(naive_preds)\n", "print(f\"RMSE naïf : {rmse_naive:.4f}\")\n", "print(f\"MAE naïf : {mae_naive:.4f}\")" ] }, { "cell_type": "markdown", "id": "02808fa9", "metadata": {}, "source": [ "Evaluation du modèle ALS obtenu :" ] }, { "cell_type": "code", "execution_count": null, "id": "86c88188", "metadata": { "hide-output": false }, "outputs": [], "source": [ "predictions = model.transform(test)\n", "\n", "rmse = evaluator_rmse.evaluate(predictions)\n", "mae = evaluator_mae.evaluate(predictions)\n", "print(f\"RMSE : {rmse:.4f}\")\n", "print(f\"MAE : {mae:.4f}\")" ] }, { "cell_type": "markdown", "id": "58fe9a4a", "metadata": {}, "source": [ "### Question\n", "\n", "L’erreur RMSE (*Root Mean Square Error*) obtenue vous semble satisfaisante ?\n", "Quelle amélioration apporte ALS par rapport au prédicteur naïf ?" ] }, { "cell_type": "markdown", "id": "4d319426", "metadata": {}, "source": [ "### Inspection des facteurs latents" ] }, { "cell_type": "code", "execution_count": null, "id": "533caaff", "metadata": { "hide-output": false }, "outputs": [], "source": [ "# Facteurs des films : chaque film est un vecteur de dimension rank\n", "item_factors = model.itemFactors\n", "item_factors.show(5, truncate=False)\n", "\n", "# Facteurs des utilisateurs\n", "user_factors = model.userFactors\n", "print(f\"Utilisateurs : {user_factors.count()}, Films : {item_factors.count()}\")" ] }, { "cell_type": "markdown", "id": "87db94ff", "metadata": {}, "source": [ "### Exercice\n", "\n", "Calculez la norme $ L_2 $ des vecteurs de films et triez par norme décroissante.\n", "A quoi correspondent les films de forte norme (films très populaires, très\n", "polarisants, etc.) ?" ] }, { "cell_type": "markdown", "id": "8de9fafa", "metadata": {}, "source": [ "### Génération de recommandations\n", "\n", "Obtenons d’abord les 10 meilleures recommandations pour chaque utilisateur :" ] }, { "cell_type": "code", "execution_count": null, "id": "69065d53", "metadata": { "hide-output": false }, "outputs": [], "source": [ "user_recs = model.recommendForAllUsers(numItems=10)\n", "user_recs.show(5, truncate=False)" ] }, { "cell_type": "markdown", "id": "30a681ff", "metadata": {}, "source": [ "Les films déjà notés par un utilisateur spécifique, identifié par son `user_id` :" ] }, { "cell_type": "code", "execution_count": null, "id": "9a462ea8", "metadata": { "hide-output": false }, "outputs": [], "source": [ "user_id = 1\n", "\n", "# Films déjà notés par cet utilisateur\n", "deja_notes = (ratings\n", " .filter(F.col(\"userId\") == user_id)\n", " .join(movies, on=\"movieId\")\n", " .orderBy(F.desc(\"rating\"))\n", " .select(\"title\", \"rating\", \"genres\"))\n", "print(f\"Films notés par l'utilisateur {user_id} :\")\n", "deja_notes.show(10, truncate=False)" ] }, { "cell_type": "markdown", "id": "fce1cd34", "metadata": {}, "source": [ "Les meilleures recommandations obtenues avec ALS pour cet utilisateur :" ] }, { "cell_type": "code", "execution_count": null, "id": "8d6929b8", "metadata": { "hide-output": false }, "outputs": [], "source": [ "recs_user = (user_recs\n", " .filter(F.col(\"userId\") == user_id)\n", " .select(F.explode(\"recommendations\").alias(\"rec\"))\n", " .select(\n", " F.col(\"rec.movieId\").alias(\"movieId\"),\n", " F.col(\"rec.rating\").alias(\"score_predit\"))\n", " .join(movies, on=\"movieId\")\n", " .select(\"title\", \"genres\", \"score_predit\")\n", " .orderBy(F.desc(\"score_predit\")))\n", "\n", "print(f\"Top 10 recommandations pour l'utilisateur {user_id} :\")\n", "recs_user.show(10, truncate=False)" ] }, { "cell_type": "markdown", "id": "ac66bd50", "metadata": {}, "source": [ "### Question\n", "\n", "Les recommandations sont-elles cohérentes avec les films déjà notés par\n", "cet utilisateur ? Trouvez-vous une cohérence de genre ou de style ?\n", "\n", "Trouvons maintenant les 10 utilisateurs les plus susceptibles d’apprécier un film\n", "spécifique, identifié par son titre :" ] }, { "cell_type": "code", "execution_count": null, "id": "4b4f7ac8", "metadata": { "hide-output": false }, "outputs": [], "source": [ "film_titre = \"Toy Story (1995)\"\n", "film_id = movies.filter(F.col(\"title\") == film_titre).first()[\"movieId\"]\n", "\n", "film_recs = model.recommendForAllItems(numUsers=10)\n", "(film_recs\n", " .filter(F.col(\"movieId\") == film_id)\n", " .select(F.explode(\"recommendations\").alias(\"rec\"))\n", " .select(F.col(\"rec.userId\").alias(\"userId\"),\n", " F.col(\"rec.rating\").alias(\"score_predit\"))\n", " .orderBy(F.desc(\"score_predit\"))\n", " .show(10))" ] }, { "cell_type": "markdown", "id": "7000b727", "metadata": {}, "source": [ "## Partie 3 : Réglage des hyperparamètres\n", "\n", "Nous explorons ici l’espace des hyperparamètres afin de voir si nos choix de\n", "départ étaient raisonnables. Cette partie peut prendre plus de temps, en attendant\n", "que les calculs soient faits regardez les questions qui suivent." ] }, { "cell_type": "markdown", "id": "e023be7f", "metadata": {}, "source": [ "### Effet du rang\n", "\n", "Le rang $ m $ est le nombre de facteurs latents. Un rang trop faible\n", "sous-ajuste alors qu’un rang trop élevé sur-ajuste (et ralentit les calculs)." ] }, { "cell_type": "code", "execution_count": null, "id": "40b7b624", "metadata": { "hide-output": false }, "outputs": [], "source": [ "resultats = []\n", "for rank in [5, 10, 20, 50]:\n", " m = ALS(rank=rank, maxIter=5, regParam=0.1,\n", " userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n", " coldStartStrategy=\"drop\")\n", " mod = m.fit(train)\n", " rmse = evaluator_rmse.evaluate(mod.transform(test))\n", " resultats.append({\"rank\": rank, \"RMSE\": rmse})\n", " print(f\"rank={rank:3d} RMSE={rmse:.4f}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "0f9c025a", "metadata": { "hide-output": false }, "outputs": [], "source": [ "res_pl = pl.DataFrame(resultats)\n", "alt.Chart(res_pl).mark_line(point=True).encode(\n", " x=alt.X(\"rank:O\", title=\"Rang (dimension latente)\"),\n", " y=alt.Y(\"RMSE:Q\", scale=alt.Scale(zero=False), title=\"RMSE test\"),\n", " tooltip=[\"rank:O\", alt.Tooltip(\"RMSE:Q\", format=\".4f\")]\n", ").properties(title=\"RMSE suivant le rang ALS\", width=400, height=280)" ] }, { "cell_type": "markdown", "id": "5e3f56d0", "metadata": {}, "source": [ "### Question\n", "\n", "La RMSE diminue systématiquement avec l’augmentation du rang ? À partir de\n", "quel rang on observe une stagnation ou une dégradation (sur-apprentissage) ?" ] }, { "cell_type": "markdown", "id": "6fec680f", "metadata": {}, "source": [ "### Effet de la régularisation" ] }, { "cell_type": "code", "execution_count": null, "id": "a6f18fb9", "metadata": { "hide-output": false }, "outputs": [], "source": [ "resultats_reg = []\n", "for reg in [0.01, 0.05, 0.1, 0.2]:\n", " m = ALS(rank=10, maxIter=5, regParam=reg,\n", " userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n", " coldStartStrategy=\"drop\")\n", " mod = m.fit(train)\n", " rmse = evaluator_rmse.evaluate(mod.transform(test))\n", " resultats_reg.append({\"regParam\": reg, \"RMSE\": rmse})\n", " print(f\"regParam={reg:.2f} RMSE={rmse:.4f}\")" ] }, { "cell_type": "markdown", "id": "d650ce6f", "metadata": {}, "source": [ "### Question\n", "\n", "Quel est l’effet d’une régularisation trop faible ? Trop forte ? Quelle\n", "valeur de `regParam` donne le meilleur RMSE ?" ] }, { "cell_type": "markdown", "id": "044c5752", "metadata": {}, "source": [ "### Recherche en grille avec validation croisée\n", "\n", "Une recherche en grille permet de couvrir de façon plus systématique l’espace de\n", "variation des valeurs des hyper-paramètres. Attention, l’exécution de cette\n", "cellule peut prendre jusqu’à 30 minutes." ] }, { "cell_type": "code", "execution_count": null, "id": "7e80b9fb", "metadata": { "hide-output": false }, "outputs": [], "source": [ "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n", "\n", "als_cv = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n", " coldStartStrategy=\"drop\")\n", "\n", "param_grid = (ParamGridBuilder()\n", " .addGrid(als_cv.rank, [20, 50])\n", " .addGrid(als_cv.regParam, [0.05, 0.1])\n", " .build())\n", "\n", "cv = CrossValidator(\n", " estimator=als_cv,\n", " estimatorParamMaps=param_grid,\n", " evaluator=evaluator_rmse,\n", " numFolds=3,\n", " parallelism=2) # évaluer 2 configurations en parallèle\n", "\n", "cv_model = cv.fit(train)\n", "print(\"Meilleure configuration :\")\n", "print(f\" rank = {cv_model.bestModel.rank}\")\n", "print(f\" regParam = {cv_model.bestModel._java_obj.parent().getRegParam()}\")\n", "print(f\" RMSE = {evaluator_rmse.evaluate(cv_model.transform(test)):.4f}\")" ] }, { "cell_type": "markdown", "id": "7dbb454c", "metadata": {}, "source": [ "### Question\n", "\n", "La validation croisée donne-t-elle une configuration différente de celle\n", "trouvée manuellement ? Comment expliquer d’éventuelles différences ?" ] }, { "cell_type": "markdown", "id": "d6347235", "metadata": {}, "source": [ "## Partie 4 : Recommandation par graphe avec GraphFrames\n", "\n", "Le filtrage collaboratif avec factorisation matricielle par ALS exploite\n", "la matrice d’utilités mais ignore la structure de réseau entre utilisateurs\n", "et articles. Nous illustrons maintenant la recommandation par *Personalized\n", "PageRank* (PPR) sur le graphe biparti qui représente les liens entre\n", "utilisateurs et films, en travaillant sur un sous-ensemble MovieLens 100K\n", "(pour une exécution plus rapide)." ] }, { "cell_type": "code", "execution_count": null, "id": "637198fc", "metadata": { "hide-output": false }, "outputs": [], "source": [ "# Sous-ensemble : MovieLens 100K, déjà employé dans tpPandasPolarsPlotly\n", "dossier100k = \"tpPandasPolarsPlotly/data/\"\n", "\n", "ratings_100k = spark.read.csv(\n", " dossier100k + \"ratings.csv\",\n", " header=True, inferSchema=True)\n", "movies_100k = spark.read.csv(\n", " dossier100k + \"movies.csv\",\n", " header=True, inferSchema=True)\n", "\n", "print(f\"Ratings 100K : {ratings_100k.count():,}\")" ] }, { "cell_type": "markdown", "id": "b911ec5f", "metadata": {}, "source": [ "### Construction du graphe biparti\n", "\n", "Dans le graphe biparti, les nœuds sont soit des utilisateurs soit des films.\n", "Nous préfixons les identifiants pour éviter les collisions. Aussi, pour que *PageRank* fonctionne il est nécessaire de rendre les liens symétriques, sinon la marche aléatoire s’arrête après le premier lien utilisateur → film. Pour cela on crée une copie de l’ensemble des arêtes, en renommant `src` en `dest` et `dest` en `src`, et ensuite on fusionne les deux ensembles. Il est critique d’utiliser `unionByName` et non `union` car `unionByName` fusionne en tenant compte du nom des colonnes alors que `union` se contente de leur position (première colonne avec première colonne, deuxième avec deuxième, etc.)." ] }, { "cell_type": "code", "execution_count": null, "id": "c5b82fc9", "metadata": { "hide-output": false }, "outputs": [], "source": [ "from graphframes import GraphFrame\n", "from pyspark.sql.types import StringType\n", "\n", "# Nœuds utilisateurs\n", "verts_users = (ratings_100k\n", " .select(\"userId\")\n", " .distinct()\n", " .withColumn(\"id\", F.concat(F.lit(\"u_\"), F.col(\"userId\").cast(StringType())))\n", " .withColumn(\"type\", F.lit(\"user\"))\n", " .select(\"id\", \"type\"))\n", "\n", "# Nœuds films\n", "verts_items = (movies_100k\n", " .select(\"movieId\", \"title\")\n", " .withColumn(\"id\", F.concat(F.lit(\"m_\"), F.col(\"movieId\").cast(StringType())))\n", " .withColumn(\"type\", F.lit(\"movie\"))\n", " .select(\"id\", F.col(\"title\").alias(\"type\")))\n", "\n", "vertices = verts_users.union(\n", " verts_items.withColumn(\"type\", F.lit(\"movie\"))\n", " .select(\"id\", \"type\"))\n", "\n", "# Arêtes utilisateur → film avec poids = note\n", "edges = (ratings_100k\n", " .withColumn(\"src\", F.concat(F.lit(\"u_\"), F.col(\"userId\").cast(StringType())))\n", " .withColumn(\"dst\", F.concat(F.lit(\"m_\"), F.col(\"movieId\").cast(StringType())))\n", " .withColumn(\"weight\", F.col(\"rating\"))\n", " .select(\"src\", \"dst\", \"weight\"))\n", "\n", "# Arêtes symétriques (graphe non orienté)\n", "edges_rev = edges.select(F.col(\"dst\").alias(\"src\"),\n", " F.col(\"src\").alias(\"dst\"),\n", " \"weight\")\n", "edges_bipartite = edges.unionByName(edges_rev)\n", "\n", "bipartite_g = GraphFrame(vertices, edges_bipartite)\n", "print(f\"Nœuds : {bipartite_g.vertices.count()}, \"\n", " f\"Arêtes : {bipartite_g.edges.count()}\")" ] }, { "cell_type": "code", "execution_count": null, "id": "638014e6", "metadata": { "hide-output": false }, "outputs": [], "source": [ "spark.sparkContext.setCheckpointDir(\"checkpoints\")" ] }, { "cell_type": "markdown", "id": "96118cb8", "metadata": {}, "source": [ "### PageRank personnalisé pour un utilisateur\n", "\n", "Obtenons les recommandations PPR pour le même utilisateur (que nous avions\n", "sélectionné plus haut). D’abord, quels sont les films qu’il a déjà vus, pour\n", "pouvoir les exclure de la sélection :" ] }, { "cell_type": "code", "execution_count": null, "id": "213a6e1c", "metadata": { "hide-output": false }, "outputs": [], "source": [ "target_user_id = 1\n", "source_id = f\"u_{target_user_id}\"\n", "\n", "# Films déjà notés par cet utilisateur (à exclure des recommandations)\n", "deja_vus = set(\n", " ratings_100k\n", " .filter(F.col(\"userId\") == target_user_id)\n", " .select(F.concat(F.lit(\"m_\"), F.col(\"movieId\").cast(StringType())).alias(\"mid\"))\n", " .rdd.flatMap(lambda x: x)\n", " .collect())\n", "\n", "print(f\"Films déjà notés par l'utilisateur {target_user_id} : {len(deja_vus)}\")" ] }, { "cell_type": "markdown", "id": "c90b8755", "metadata": {}, "source": [ "Ensuite les recommandations, obtenues par PPR et excluant les films déjà vus." ] }, { "cell_type": "code", "execution_count": null, "id": "72e6eadf", "metadata": { "hide-output": false }, "outputs": [], "source": [ "# PPR depuis l'utilisateur cible\n", "ppr = bipartite_g.pageRank(\n", " resetProbability=0.15,\n", " maxIter=10,\n", " sourceId=source_id)\n", "\n", "# Recommandations : films à fort score PPR non encore vus\n", "recs_ppr = (ppr.vertices\n", " .filter(F.col(\"id\").startswith(\"m_\")) # garder uniquement les films\n", " .filter(~F.col(\"id\").isin(deja_vus)) # exclure les films déjà vus\n", " .withColumn(\"movieId\",\n", " F.col(\"id\").substr(3, 10).cast(\"int\"))\n", " .join(movies_100k.select(\"movieId\", \"title\", \"genres\"), on=\"movieId\")\n", " .select(\"title\", \"genres\", \"pagerank\")\n", " .orderBy(F.desc(\"pagerank\"))\n", " .limit(10))\n", "\n", "print(f\"Top 10 recommandations PPR pour l'utilisateur {target_user_id} :\")\n", "recs_ppr.show(truncate=False)" ] }, { "cell_type": "markdown", "id": "9d2fb4a3", "metadata": {}, "source": [ "### Question\n", "\n", "Comparez les recommandations PPR avec celles d’ALS pour le même utilisateur\n", "(sachant toutefois que ALS a été entraîné sur un ensemble bien plus grand).\n", "Quelles différences observez-vous ?" ] }, { "cell_type": "markdown", "id": "58c90335", "metadata": {}, "source": [ "### Question\n", "\n", "Le PPR exploite les chemins de longueur supérieure à 2 dans le graphe\n", "biparti. Qu’est-ce que cela signifie concrètement ? Quel type d’information\n", "est ignoré par le filtrage collaboratif *user-based* au premier degré mais\n", "exploité par PPR ?" ] }, { "cell_type": "markdown", "id": "515a6982", "metadata": {}, "source": [ "### Recommandation basée sur les communautés" ] }, { "cell_type": "code", "execution_count": null, "id": "dcb63c74", "metadata": { "hide-output": false }, "outputs": [], "source": [ "# Détection de communautés d'utilisateurs sur le graphe biparti\n", "lp = bipartite_g.labelPropagation(maxIter=5)\n", "n_communities_lp = lp.select(\"label\").distinct().count()\n", "print(f\"Nombre de communautés (Label Propagation) : {n_communities_lp}\")\n", "\n", "# A quelle communauté appartient l'utilisateur cible ?\n", "comm_user = lp.filter(F.col(\"id\") == source_id).first()[\"label\"]\n", "print(f\"Communauté de l'utilisateur {target_user_id} : {comm_user}\")\n", "\n", "# Films les plus populaires dans cette communauté, non encore vus par la cible\n", "users_same_comm = (lp\n", " .filter(F.col(\"label\") == comm_user)\n", " .filter(F.col(\"id\").startswith(\"u_\"))\n", " .withColumn(\"userId\",\n", " F.col(\"id\").substr(3, 10).cast(\"int\"))\n", " .select(\"userId\"))\n", "\n", "recs_comm = (ratings_100k\n", " .join(users_same_comm, on=\"userId\")\n", " .groupBy(\"movieId\")\n", " .agg(F.mean(\"rating\").alias(\"note_moy_comm\"),\n", " F.count(\"*\").alias(\"nb_notes_comm\"))\n", " .filter(F.col(\"nb_notes_comm\") >= 5)\n", " .filter(~F.concat(F.lit(\"m_\"),\n", " F.col(\"movieId\").cast(StringType())).isin(deja_vus))\n", " .join(movies_100k.select(\"movieId\", \"title\", \"genres\"), on=\"movieId\")\n", " .orderBy(F.desc(\"note_moy_comm\"))\n", " .select(\"title\", \"genres\", \"note_moy_comm\", \"nb_notes_comm\")\n", " .limit(10))\n", "\n", "print(\"Top 10 films populaires dans la communauté :\")\n", "recs_comm.show(truncate=False)" ] }, { "cell_type": "markdown", "id": "11697caa", "metadata": {}, "source": [ "### Question\n", "\n", "La recommandation par communautés vous semble pertinente ? Calculez et\n", "affichez les tailles des communautés. Que contatez-vous ?" ] }, { "cell_type": "markdown", "id": "30f63db1", "metadata": {}, "source": [ "## Synthèse\n", "\n", "Dans ce TP nous avons illustré trois approches complémentaires de recommandation :\n", "\n", "1. Le filtrage collaboratif par factorisation matricielle avec ALS (*Alternating\n", "Least Squares*) est le plus performant en termes de RMSE sur les\n", "notes explicites. Il est scalable (Spark parallélise les étapes ALS) mais\n", "ne tient pas compte de la structure de réseau et souffre du problème de\n", "démarrage à froid.\n", "\n", "2. La recommandation par graphe avec PPR (*Personalized PageRank*)\n", "exploite les chemins de longueur arbitraire\n", "dans le graphe biparti. Il est interprétable (on peut tracer le chemin menant\n", "à une recommandation) et gère naturellement les utilisateurs sans beaucoup\n", "d’historique (quelques interactions suffisent à amorcer la diffusion du score).\n", "Il est plus lent qu’ALS sur de très grands graphes.\n", "\n", "3. Recommandation par communauté : la plus simple et la plus interprétable,\n", "mais aussi la plus grossière car tous les utilisateurs d’une même communauté\n", "reçoivent les mêmes recommandations. Elle est utile comme *baseline* ou en\n", "complément des deux approches précédentes.\n", "\n", "En production, les systèmes contemporains sont **hybrides** : ALS (ou LightGCN)\n", "pour la précision, PPR ou similarité de contenu pour la gestion du démarrage\n", "à froid et filtrage par communauté pour la diversité." ] }, { "cell_type": "markdown", "id": "ce844446", "metadata": {}, "source": [ "## Références\n", "\n", "[HK15] Harper, F. M. and Konstan, J. A. *The MovieLens Datasets: History and Context*.\n", "ACM Transactions on Interactive Intelligent Systems, 5(4), 2015.\n", "[https://grouplens.org/datasets/movielens/](https://grouplens.org/datasets/movielens/)\n", "\n", "[KBV09] Koren, Y., Bell, R., Volinsky, C. *Matrix factorization techniques for\n", "recommender systems*. Computer, 42(8):30–37, 2009.\n", "\n", "[HDW20] He, X., Deng, K., Wang, X., Li, Y., Zhang, Y., Wang, M. *LightGCN: Simplifying\n", "and Powering Graph Convolution Network for Recommendation*. SIGIR 2020.\n", "\n", "Documentation Spark MLlib ALS :\n", "[https://spark.apache.org/docs/latest/ml-collaborative-filtering.html](https://spark.apache.org/docs/latest/ml-collaborative-filtering.html)" ] } ], "metadata": { "date": 1790953173.559263, "filename": "tpRecommandation.rst", "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python" }, "title": "Travaux pratiques - Recommandation avec Spark MLlib et GraphFrames" }, "nbformat": 4, "nbformat_minor": 5 }