{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.14","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":31254,"databundleVersionId":3103714,"sourceType":"competition"},{"sourceId":4508,"sourceType":"datasetVersion","datasetId":138},{"sourceId":77759,"sourceType":"datasetVersion","datasetId":339},{"sourceId":9841804,"sourceType":"datasetVersion","datasetId":6037753}],"dockerImageVersionId":30786,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Lab8 Recommendation with PySpark\n```\nSource：\nSpark The Definitive Guide Chapter 28. Recommendation\n```","metadata":{}},{"cell_type":"code","source":"# Install pyspark\n!pip install pyspark","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:11:57.843451Z","iopub.execute_input":"2024-11-25T07:11:57.843855Z","iopub.status.idle":"2024-11-25T07:12:45.413108Z","shell.execute_reply.started":"2024-11-25T07:11:57.843816Z","shell.execute_reply":"2024-11-25T07:12:45.411872Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Create a spark session\nfrom pyspark.sql import SparkSession\nspark = SparkSession.builder.config(\"spark.sql.pivotMaxValues\", \"30000\").getOrCreate()\nspark.sparkContext.setLogLevel(\"ERROR\")\nspark ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:12:45.415576Z","iopub.execute_input":"2024-11-25T07:12:45.416192Z","iopub.status.idle":"2024-11-25T07:12:53.803868Z","shell.execute_reply.started":"2024-11-25T07:12:45.416138Z","shell.execute_reply":"2024-11-25T07:12:53.802366Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 基于内容的过滤","metadata":{}},{"cell_type":"markdown","source":"## Load data","metadata":{}},{"cell_type":"code","source":"# Load dataset\nfile_path = \"/kaggle/input/tmdb-movie-metadata/tmdb_5000_movies.csv\"  \ndf = spark.read.option(\"escape\", '\"').option(\"header\", True).option(\"inferSchema\", True).csv(file_path).dropna('any')\n\n# Initial data examination\ndf.select(\"title\",\"genres\",\"overview\").show(2,False,True)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:12:53.806031Z","iopub.execute_input":"2024-11-25T07:12:53.806852Z","iopub.status.idle":"2024-11-25T07:13:05.015091Z","shell.execute_reply.started":"2024-11-25T07:12:53.806793Z","shell.execute_reply":"2024-11-25T07:13:05.014118Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Process features\n1. #### Convert genre to binary feaures\n1. #### Convert overview to TF-IDF features\n1. #### Combine genre and overview ","metadata":{}},{"cell_type":"code","source":"# Process genre\nfrom pyspark.sql.types import *\nfrom pyspark.sql.functions import *\n\nlist_schema = ArrayType(StructType([\n  StructField(\"id\", StringType(), True),\n  StructField(\"name\", StringType(), True)\n]))\n\ncols = 'genres'\n\ngenre = df.withColumn('genre',explode(from_json(cols, list_schema)))\\\n.withColumn('genre',col('genre.name'))\\\n.groupBy('title')\\\n.pivot('genre')\\\n.agg(lit(1)).na.fill(0)\n\nfrom pyspark.ml.feature import VectorAssembler\nva = VectorAssembler().setInputCols(genre.columns[1:-1]).setOutputCol('genreVec')\ngenreVec = va.transform(genre).select('title', 'genreVec')\ngenreVec.show(5, False)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:05.017480Z","iopub.execute_input":"2024-11-25T07:13:05.017928Z","iopub.status.idle":"2024-11-25T07:13:12.408976Z","shell.execute_reply.started":"2024-11-25T07:13:05.017864Z","shell.execute_reply":"2024-11-25T07:13:12.402803Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Process overview\nfrom pyspark.ml.feature import Tokenizer, StopWordsRemover, HashingTF, IDF\ntokenizer = Tokenizer(inputCol=\"overview\", outputCol=\"words\")\nremover = StopWordsRemover(inputCol=\"words\", outputCol=\"filtered_words\")\nhashing_tf = HashingTF(inputCol=\"filtered_words\", outputCol=\"raw_features\", numFeatures=1000)\nidf = IDF(inputCol=\"raw_features\", outputCol=\"tfidf_features\")\n\n# 创建 Pipeline\nfrom pyspark.ml.pipeline import Pipeline\npipeline = Pipeline(stages=[tokenizer,remover,hashing_tf,idf])\n\n# 拟合数据并转换\nmodel = pipeline.fit(df)\ndf_tfidf = model.transform(df)\n\n# Display the resulting TF-IDF features for inspection\ndf_tfidf.select(\"title\", \"tfidf_features\").show(2,False,True)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:35.535760Z","iopub.execute_input":"2024-11-25T07:13:35.536712Z","iopub.status.idle":"2024-11-25T07:13:40.616419Z","shell.execute_reply.started":"2024-11-25T07:13:35.536666Z","shell.execute_reply":"2024-11-25T07:13:40.614522Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Combine features\ndf_cf = df_tfidf.select(\"title\", \"tfidf_features\").join(genreVec,\"title\",\"inner\")\n\n# VectorAssembler\nfrom pyspark.ml.feature import VectorAssembler\nvA = VectorAssembler(handleInvalid = 'skip' )\\\n.setInputCols([\"tfidf_features\",\"genreVec\"]).setOutputCol('features')\n\ndf_features = vA.transform(df_cf)\ndf_features.show(2)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:40.619184Z","iopub.execute_input":"2024-11-25T07:13:40.619720Z","iopub.status.idle":"2024-11-25T07:13:42.687748Z","shell.execute_reply.started":"2024-11-25T07:13:40.619666Z","shell.execute_reply":"2024-11-25T07:13:42.686010Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Calculate similarity matrix based on features","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import udf\nfrom pyspark.sql.types import DoubleType\n\n# Define the cosine similarity function\ndef cosine_similarity(v1, v2):\n    dot_product = float(v1.dot(v2))   # Calculate dot product of SparseVector v1 & v2\n    norm_v1 = float(v1.norm(2))       # Find norm of the SparseVector v1\n    norm_v2 = float(v2.norm(2))       # Find norm of the SparseVector v2\n\n    # Handle Zero Division cases\n    if norm_v1 * norm_v2 == 0:\n        return 0.0\n\n    similarity = dot_product / (norm_v1 * norm_v2)   # Calculate cosine similarity\n    return similarity\n\n# Register the function as a UDF\ncosine_similarity_udf = udf(cosine_similarity, DoubleType())\n\n# Generate similarity DataFrame using a cross join for each movie pair\ndf_cross = df_features.alias(\"a\").crossJoin(df_features.alias(\"b\")) \\\n           .withColumn(\"similarity\", cosine_similarity_udf(\"a.features\", \"b.features\")) \\\n           .select(col(\"a.title\").alias(\"movie1\"), col(\"b.title\").alias(\"movie2\"), \"similarity\") \\\n           .filter(\"movie1 != movie2\")  # Exclude self-comparisons\n\n# Show sample similarities\ndf_cross.show(5, False)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:42.694689Z","iopub.execute_input":"2024-11-25T07:13:42.695256Z","iopub.status.idle":"2024-11-25T07:13:47.871237Z","shell.execute_reply.started":"2024-11-25T07:13:42.695199Z","shell.execute_reply":"2024-11-25T07:13:47.870062Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Make recommendation based on similarity","metadata":{}},{"cell_type":"code","source":"# Recommendation function\ndef get_recommendations(title, similarity_df, n=10):\n    # Filter for the input movie in the similarity DataFrame\n    recommended = similarity_df.filter(col(\"movie1\") == title) \\\n                               .orderBy(col(\"similarity\").desc()) \\\n                               .limit(n)\n    \n    # Select movie2 as recommendations with similarity score\n    recommendations = recommended.select(\"movie2\", \"similarity\").withColumnRenamed(\"movie2\", \"Recommended Movie\")\n    \n    return recommendations","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:47.873077Z","iopub.execute_input":"2024-11-25T07:13:47.873503Z","iopub.status.idle":"2024-11-25T07:13:47.903336Z","shell.execute_reply.started":"2024-11-25T07:13:47.873457Z","shell.execute_reply":"2024-11-25T07:13:47.900311Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Testing the function with a sample movie title\nsample_title = \"Frozen\"\nrecommendations = get_recommendations(sample_title, df_cross)\nrecommendations.show(10, False)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:13:50.623356Z","iopub.execute_input":"2024-11-25T07:13:50.623804Z","iopub.status.idle":"2024-11-25T07:13:56.111576Z","shell.execute_reply.started":"2024-11-25T07:13:50.623756Z","shell.execute_reply":"2024-11-25T07:13:56.109677Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 创建产品列表\nmovie_titles = [\"Frozen\", \"Stardust\",\"Paddington\"]\n\n# 初始化一个空白的结果\ncombined_recommendations = None\n\n# 遍历产品列表\nfor title in movie_titles:\n    recommendations = get_recommendations(title, df_cross)\n    #合并推荐结果\n    recommendations = recommendations.groupby().agg(collect_list(\"Recommended Movie\"))\n    #标注目标产品 \n    recommendations = recommendations.withColumn(\"source_title\", lit(title))\n    # 组合结果\n    if combined_recommendations is None:\n        combined_recommendations = recommendations\n    else:\n        combined_recommendations = combined_recommendations.union(recommendations)\n\n#展示结果\ncombined_recommendations.show(len(movie_titles), False, True)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T07:21:33.624080Z","iopub.execute_input":"2024-11-25T07:21:33.624507Z","iopub.status.idle":"2024-11-25T07:21:39.214348Z","shell.execute_reply.started":"2024-11-25T07:21:33.624470Z","shell.execute_reply":"2024-11-25T07:21:39.209221Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 练习1\n1. 请在电影类别、描述的基础上，添加budget|popularity|vote_average|revenue作为基础特征，创建基于内容的推荐系统。\n2. 查看电影\"Frozen\"的前10个推荐电影是否改变。","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 基于用户的协同过滤","metadata":{}},{"cell_type":"markdown","source":"## Load data","metadata":{}},{"cell_type":"code","source":"# Import necessary PySpark libraries\nfrom pyspark.sql.functions import *\nfrom pyspark.sql.types import *\ndf = spark.read.text(\"/kaggle/input/sample-movielens-ratings/sample_movielens_ratings.txt\")\\\n  .rdd.toDF()\\\n  .selectExpr(\"split(value , '::') as col\")\\\n  .selectExpr(\n    \"cast(col[0] as int) as userId\",\n    \"cast(col[1] as int) as movieId\",\n    \"cast(col[2] as float) as rating\",\n    \"cast(col[3] as long) as timestamp\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df.show(5,False)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df.select(count_distinct(\"userId\")).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df.select(count_distinct(\"movieId\")).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 创建用户评分矩阵\n# 将数据按用户和电影的评分聚合，并填充空值\n#ratings_matrix = df.groupBy(\"userId\").pivot(\"movieId\").agg(first(\"rating\")).na.fill(0)\n\n# ##修改## 以适配columnSimilarity，因为columnSimilarity计算的是列与列之间的相似性，因此用户需要是用列代表\nratings_matrix = df.sort(\"userId\").groupBy(\"movieId\").pivot(\"userId\").agg(first(\"rating\")).na.fill(0)\nratings_matrix.select(ratings_matrix.columns[0:6]).show(5)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Cosine similarity","metadata":{}},{"cell_type":"code","source":"# ##修改## 以简化步骤\n\n# 将数据按用户和电影的评分聚合，并填充空值\n#因为columnSimilarity计算的是列与列之间的相似性，因此用户需要是用列代表\nratings_matrix = df.sort(\"userId\").groupBy(\"movieId\").pivot(\"userId\").agg(first(\"rating\")).na.fill(0)\n\n# 转换 DataFrame 为 RDD\nfrom pyspark.ml.linalg import DenseVector\nrdd = ratings_matrix.drop(\"movieId\").rdd.map(lambda row: list(row)) \n\n# 创建 RowMatrix\nfrom pyspark.mllib.linalg.distributed import RowMatrix\nrow_matrix = RowMatrix(rdd)\n\n# 算用户相似度（列相似度）\nsimilarity_matrix = row_matrix.columnSimilarities()\n\n# 转换结果为 DataFrame\nuser_similarity_df = similarity_matrix.entries.toDF([\"userId1\", \"userId2\", \"similarity\"])\n\n# 显示结果\nuser_similarity_df.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# # Step 1: 将评分矩阵转换为特征向量\n# from pyspark.ml.feature import VectorAssembler\n# assembler = VectorAssembler(inputCols=ratings_matrix.columns[1:], outputCol=\"features\")\n# ratings_vector_df = assembler.transform(ratings_matrix).select(\"userId\", \"features\")\n# ratings_vector_df.show(5)\n# # Step 2: 将 DataFrame 转换为 RDD，并创建 RowMatrix,用于计算相似度\n# # 提取评分向量并转换为 MLlib 的 RowMatrix 格式\n# from pyspark.mllib.linalg.distributed import RowMatrix\n# ratings_rdd = ratings_vector_df.select(\"features\").rdd.map(lambda row: row[0].toArray())  # 转换为 RDD\n# row_matrix = RowMatrix(ratings_rdd)\n\n# # Step 3: 计算用户之间的余弦相似度\n# similarity_matrix = row_matrix.columnSimilarities()\n\n# # Step 4: 将相似度矩阵转换为 DataFrame 以便查看结果\n# user_similarity_df = similarity_matrix.entries.toDF([\"userId1\", \"userId2\", \"similarity\"])\n\n# # 显示前 10 个用户相似度\n# user_similarity_df.show(10)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Generate recommendations","metadata":{}},{"cell_type":"code","source":"# Generate recommendations by finding movies highly rated by similar users\n# Join user similarity with the original ratings to get similar users' ratings\nrecommendations_df = user_similarity_df.join(df, user_similarity_df.userId2 == df.userId)\nrecommendations_df.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"recommendations_df.orderBy(col(\"userId1\"),col('movieId'),col('similarity').desc()).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Calculate a weighted score for recommendations (similarity * rating)\nrecommendations_df = recommendations_df.withColumn(\"weighted_score\", col(\"similarity\") * col(\"rating\"))\\\n.groupBy(\"userId1\", \"movieId\")\\\n.agg(sum(\"weighted_score\").alias(\"total_score\"))\nrecommendations_df.orderBy(col(\"userId1\"),col('movieId')).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"recommendations_df.orderBy(col(\"userId1\"),col(\"total_score\").desc()).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 使用窗口函数按每个用户排序，并获取前5个推荐\nfrom pyspark.sql.window import Window\nwindowSpec = Window.partitionBy(\"userId1\").orderBy(col(\"total_score\").desc())\nrecommended_movies_df = recommendations_df.withColumn(\"rank\", row_number().over(windowSpec)) \\\n    .filter(col(\"rank\") <= 5)  # 仅保留前5个推荐\n\n# 显示每个用户的前5个推荐电影\nrecommended_movies_df.select(\"userId1\", \"movieId\", \"total_score\", \"rank\").show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 练习2：\n1. 请尝试使用/kaggle/input/movielens-20m-dataset数据，抽取其中一个小样本，创建一个基于用户的推荐系统.\n2. 选择5个用户，向他们各自推荐5部电影，列出推荐的片名。","metadata":{}},{"cell_type":"code","source":"# Load the datasets \nmovie_path = \"/kaggle/input/movielens-20m-dataset/movie.csv\"  \nrating_path = \"/kaggle/input/movielens-20m-dataset/rating.csv\"  \nmovie_df = spark.read.csv(movie_path, header=True, inferSchema=True).limit(1000)\nrating_df = spark.read.csv(rating_path, header=True, inferSchema=True)\n\n# List of user IDs to keep\nuser_subset = rating_df.select(\"userId\").distinct().limit(1000).rdd.flatMap(lambda x: x).collect()\n\n# Filter DataFrame to keep only the specified users\nrating_df = rating_df.filter(col(\"userId\").isin(user_subset))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 基于模型的协同过滤","metadata":{"execution":{"iopub.status.busy":"2024-11-08T08:32:27.936115Z","iopub.execute_input":"2024-11-08T08:32:27.937028Z","iopub.status.idle":"2024-11-08T08:32:27.952155Z","shell.execute_reply.started":"2024-11-08T08:32:27.936933Z","shell.execute_reply":"2024-11-08T08:32:27.950125Z"}}},{"cell_type":"markdown","source":"## 探索性分析","metadata":{}},{"cell_type":"code","source":"# Row-based format\ndata = spark.read.csv('../input/movielens-20m-dataset/rating.csv',\n                      inferSchema=True,\n                      header=True)\ndata.show(5)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Userid, movieid: integer\ndata.printSchema()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Random sampling\nratings = data.sample(0.01,1234).cache()\nprint(ratings.count())\ndata.unpersist()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Sparsity\n# Number of ratings in matrix\nnumerator = ratings.count()\n\n# Distinct users and movies\nusers = ratings.select(\"userId\").distinct().count()\nmovies = ratings.select(\"movieId\").distinct().count()\n\n# Number of ratings matrix could contain if no empty cells\ndenominator = users * movies\n\n#Calculating sparsity\nsparsity = 1 - (numerator*1.0 / denominator)\nprint (\"Sparsity: \", sparsity)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Distribution\n# Count ratings for each user\nuser_rating_counts = ratings.groupBy(\"userId\").count()\n\n# Get descriptive statistics for user rating counts\nuser_rating_counts.describe().show()\n\n# Count ratings for each movie\nmovie_rating_counts = ratings.groupBy(\"movieId\").count()\n\n# Get descriptive statistics for movie rating counts\nmovie_rating_counts.describe().show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 模型训练\n### ALS基本模型","metadata":{}},{"cell_type":"code","source":"%%time\n# Split data\n(training_data, test_data) = ratings.randomSplit([0.5, 0.5], 1234)\n\n# Build ALS model\nfrom pyspark.ml.recommendation import ALS\nals = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n          rank=5, maxIter=10,regParam=.05, \n          nonnegative=True,\n          coldStartStrategy=\"drop\", \n          implicitPrefs=False)\n\n# Fit model to training data\nmodel = als.fit(training_data)\n\n# Generate predictions on test_data\npredictions = model.transform(test_data)\n\n# Tell Spark how to evaluate predictions\nfrom pyspark.ml.evaluation import RegressionEvaluator\nevaluator = RegressionEvaluator(\n    metricName=\"rmse\", \n    labelCol=\"rating\",\n    predictionCol=\"prediction\")\n\n# Obtain and print RMSE\nrmse = evaluator.evaluate(predictions)\nprint(\"RMSE: \",rmse) ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 交叉验证模型","metadata":{}},{"cell_type":"code","source":"%%time\n# Split data\n(training_data, test_data) = ratings.randomSplit([0.5, 0.5], 1234)\n\n# Build generic ALS model without hyperparameters\nals = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n          coldStartStrategy=\"drop\", nonnegative = True,\n          implicitPrefs = False)\n\n# Tell Spark what values to try for each hyperparameter\nfrom pyspark.ml.tuning import ParamGridBuilder\nparam_grid = ParamGridBuilder()\\\n.addGrid(als.rank, [5, 120])\\\n.addGrid(als.maxIter, [5, 10])\\\n.addGrid(als.regParam, [.05, 1.5])\\\n.build()\n\n# Tell Spark how to evaluate model performance\nevaluator = RegressionEvaluator(metricName=\"rmse\", \n                                labelCol=\"rating\",\n                                predictionCol=\"prediction\")\n\n# Build cross validation step using CrossValidator\nfrom pyspark.ml.tuning import CrossValidator\ncv = CrossValidator(estimator = als,\n                    estimatorParamMaps = param_grid,\n                    evaluator = evaluator,\n                    numFolds = 3)\n\n# Run the cv on the training data\nmodel = cv.fit(training_data)\n\n# [Stage 2469:============================================>          (8 + 2) / 10]\n# CPU times: user 1.9 s, sys: 462 ms, total: 2.37 s\n# Wall time: 9min 54s","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 模型预测与评估","metadata":{}},{"cell_type":"code","source":"# Extract best combination of values from cross validation\nbest_model = model.bestModel\n\n# Generate test set predictions and evaluate using RMSE\npredictions = best_model.transform(test_data)\nrmse = evaluator.evaluate(predictions)\n\n# Print evaluation metrics and model parameters\nprint(\"**Best Model**\")\nprint(\"RMSE = \", rmse)\nprint(\" Rank: \", best_model.rank)\nprint(\" MaxIter: \", best_model._java_obj.parent().getMaxIter())\nprint(\" RegParam: \", best_model._java_obj.parent().getRegParam())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 模型应用：推荐结果","metadata":{}},{"cell_type":"code","source":"best_model.recommendForAllUsers(3)\\\n.selectExpr(\"userId\", \"explode(recommendations) as c\")\\\n.withColumn(\"movieId\",col('c').movieID)\\\n.printSchema()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Recommend 3 movies to each user \nrec3 = best_model.recommendForAllUsers(3)\\\n.selectExpr(\"userId\", \"explode(recommendations) as c\")\\\n.withColumn(\"movieId\",col('c').movieID)\\\n.withColumn(\"rating\",col('c').rating)\\\n.drop(col('c'))\nrec3.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Recommend 3 users to each movie \nrec3i = best_model.recommendForAllItems(3)\\\n.selectExpr(\"movieId\", \"explode(recommendations) as c\")\\\n.withColumn(\"userId\",col('c').userId)\\\n.withColumn(\"rating\",col('c').rating)\\\n.drop(col('c'))\nrec3i.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Movie names \nmovie = spark.read.csv('../input/movielens-20m-dataset/movie.csv',\n                      inferSchema=True,\n                      header=True)\nmovie.printSchema()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"rec3_name = rec3.join(movie, \"movieId\", 'inner')\nrec3_name.show(20, False)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 练习3：最优电影推荐系统\n1. 请根据MovieLens数据重新设计一个电影推荐系统，设置更多的参数组合，通过交叉验证模型训练一个最优模型。最优模型的参数是什么？模型的RMSE是多少？\n2. 请根据MovieLens数据主页的介绍了解数据，结合机器学习、数据挖掘等课程的内容、以及分布式大数据分析与推荐系统的理论知识，思考以电影推荐系统为主题进行毕业设计（毕业论文）的可行性。","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 练习4：H&M 推荐系统\n1. 请使用/kaggle/input/h-and-m-personalized-fashion-recommendations中的数据，创建一个基于模型的服装推荐系统。\n2. 使用交叉验证模型选取最优模型。最优模型的超参分别是什么？RMSE是多少？\n3. 选择10个用户，应用模型向每个用户推荐10个商品。","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null}]}