{"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"}],"dockerImageVersionId":30786,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# 《实验报告5：PySpark推荐系统》\n---\n### 实验内容：\n1. 创建基于内容的推荐系统\n2. 创建基于用户的协同过滤推荐系统\n3. 使用Spark MLlib创建基于模型的推荐系统。\n\n### 实验目标：\n1. 学会余弦相似度的计算\n2. 掌握Spark MLlib ALS算法。\n3. 练习陌生数据的预处理与探索性分析。\n4. 复习PySpark的特征工程工具、管道工具、交叉验证工具。\n\n### 实验提示:\n\n\n### 实验报告作业提交要求: \n### 本次实验为个人作业，每位同学须单独完成并提交以下文档\n1. 实验报告提交清单\n   - （1）**完成的PDF格式的Jupyter Notebook + 填写好的实验报告纸**：\n       - 请用浏览器的打印功能，将本Jupyter Notebook打印为PDF文档，命名为　**组号-学号-姓名-实验报告5.pdf**，提交到相应的超星平台作业提交处。\n       - 提交的PDF文档需要完整显示（１）问题、（2）你输入的代码、（3）运行代码输出的结果、（4）对问题的回答。\n       - 由于可存档的PDF文件是实验报告归档的必要内容，实验作业将根据PDF显示的结果打分，**未提交PDF文档本次实验0分**。\n       - 请将实验报告纸模板作为该文档的封面，无法合并PDF的小组，可以分开提交两份PDF，并将实验报告纸模板命名为　**组号-学号-姓名-实验报告5-封面.pdf**。\n   - （2）**完成的ipynb格式的Jupyter Notebook**：请同时提交ipynb格式的完成的Jupyter Notebook到相应的超星平台作业提交处，以作代码相似度分析，**未提交ipynb文档本次实验0分**。\n2. 在每一题中，请根据要求填写代码，或将不完整的代码补充完整，并回答相应问题。本次实验总分**22**分。","metadata":{"_kg_hide-output":true}},{"cell_type":"markdown","source":"# H&M推荐系统\n1. 请对/kaggle/input/h-and-m-personalized-fashion-recommendations中的数据进行探索性分析。由于数据较大，注意进行数据抽样，选取一个数据子集进行探索和建模。","metadata":{}},{"cell_type":"code","source":"!pip install pyspark\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col, udf, expr, lit, rand\nfrom pyspark.sql.types import DoubleType, IntegerType, StringType, ArrayType\nfrom pyspark.ml.feature import CountVectorizer\nfrom pyspark.ml.recommendation import ALS\nfrom pyspark.ml.evaluation import RegressionEvaluator\nfrom pyspark.ml.tuning import CrossValidator, ParamGridBuilder","metadata":{"execution":{"iopub.status.busy":"2024-11-25T06:26:01.199253Z","iopub.execute_input":"2024-11-25T06:26:01.199626Z","iopub.status.idle":"2024-11-25T06:26:12.384977Z","shell.execute_reply.started":"2024-11-25T06:26:01.199584Z","shell.execute_reply":"2024-11-25T06:26:12.383569Z"},"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col, lit, when, udf, collect_list, array\nfrom pyspark.ml.feature import CountVectorizer, StringIndexer, VectorAssembler\nfrom pyspark.ml.recommendation import ALS\nfrom pyspark.ml.evaluation import RegressionEvaluator\nfrom pyspark.ml.tuning import ParamGridBuilder, CrossValidator\nimport random\nspark = SparkSession.builder \\\n    .appName(\"H&M Recommendation System\") \\\n    .getOrCreate()\narticles = spark.read.csv('/kaggle/input/h-and-m-personalized-fashion-recommendations/articles.csv', header=True, inferSchema=True)\ncustomers = spark.read.csv('/kaggle/input/h-and-m-personalized-fashion-recommendations/customers.csv', header=True, inferSchema=True)\ntransactions = spark.read.csv('/kaggle/input/h-and-m-personalized-fashion-recommendations/transactions_train.csv', header=True, inferSchema=True)\ntransactions.show(5)\narticles.show(5)\ncustomers.show(5)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:26:25.303816Z","iopub.execute_input":"2024-11-25T06:26:25.304709Z","iopub.status.idle":"2024-11-25T06:27:17.156156Z","shell.execute_reply.started":"2024-11-25T06:26:25.304664Z","shell.execute_reply":"2024-11-25T06:27:17.155049Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"2. 请以你的学号为随机种子，随机选取任意10个用户的ID，创建一个名为user_list的python list。","metadata":{}},{"cell_type":"code","source":"user_list = sample_transactions_df.select(\"customer_id\").distinct().orderBy(rand(seed=2211020111)).limit(10)\nuser_list = [row.customer_id for row in user_list.collect()]\nprint(\"User List:\", user_list)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T05:40:29.223638Z","iopub.execute_input":"2024-11-25T05:40:29.224489Z","iopub.status.idle":"2024-11-25T05:40:58.198635Z","shell.execute_reply.started":"2024-11-25T05:40:29.224443Z","shell.execute_reply":"2024-11-25T05:40:58.197722Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"3. 创建一个基于内容的推荐系统。","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.functions import col\nfrom pyspark.sql.types import DoubleType\nfrom pyspark.sql.functions import udf\nimport numpy as np\n\ntarget_article_id = 505221004\ntarget_features_row = articles_features_df.filter(col(\"article_id\") == target_article_id).select(\"features\").first()\n\nif target_features_row:\n    target_features = np.array(target_features_row.features.toArray())\n\n    def cosine_similarity(features):\n        if features is None:\n            return 0.0\n        features = np.array(features.toArray())\n        dot_product = np.dot(features, target_features)\n        norm_features = np.linalg.norm(features)\n        norm_target = np.linalg.norm(target_features)\n        return float(dot_product / (norm_features * norm_target)) if norm_features > 0 and norm_target > 0 else 0.0\n\n    cosine_similarity_udf = udf(cosine_similarity, DoubleType())\n\n    similarity_df = articles_features_df.withColumn(\n        \"similarity\", cosine_similarity_udf(col(\"features\"))\n    ).orderBy(col(\"similarity\").desc())\n\n    similarity_df.select(\"article_id\", \"product_type_name\", \"similarity\").show(3)\nelse:\n    print(f\"Target article_id {target_article_id} not found.\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-21T16:44:15.838397Z","iopub.execute_input":"2024-11-21T16:44:15.83888Z","iopub.status.idle":"2024-11-21T16:44:24.313101Z","shell.execute_reply.started":"2024-11-21T16:44:15.838838Z","shell.execute_reply":"2024-11-21T16:44:24.311872Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"4. 用上述推荐系统，向user_list中的用户各推荐3个产品，列出产品的名称和类别。","metadata":{}},{"cell_type":"code","source":"target_article_id = 685687004\ntarget_features_row = articles_features_df.filter(col(\"article_id\") == target_article_id).select(\"features\").first()\n\nif target_features_row:\n    target_features = np.array(target_features_row.features.toArray())\n\n    def cosine_similarity(features):\n        if features is None:\n            return 0.0\n        features = np.array(features.toArray())\n        dot_product = np.dot(features, target_features)\n        norm_features = np.linalg.norm(features)\n        norm_target = np.linalg.norm(target_features)\n        return float(dot_product / (norm_features * norm_target)) if norm_features > 0 and norm_target > 0 else 0.0\n\n    cosine_similarity_udf = udf(cosine_similarity, DoubleType())\n\n    similarity_df = articles_features_df.withColumn(\n        \"similarity\", cosine_similarity_udf(col(\"features\"))\n    ).orderBy(col(\"similarity\").desc())\n\n    similarity_df.select(\"article_id\", \"product_type_name\", \"similarity\").show(3)\nelse:\n     print(f\"Target article_id {target_article_id} not found.\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-21T16:52:00.911498Z","iopub.execute_input":"2024-11-21T16:52:00.912819Z","iopub.status.idle":"2024-11-21T16:52:07.067291Z","shell.execute_reply.started":"2024-11-21T16:52:00.912765Z","shell.execute_reply":"2024-11-21T16:52:07.066122Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"5. 创建一个基于用户的协同过滤推荐系统。","metadata":{}},{"cell_type":"code","source":"spark.stop()\n\nspark = SparkSession.builder \\\n    .appName(\"ALS Model\") \\\n    .config(\"spark.kryoserializer.buffer\", \"128m\") \\\n    .config(\"spark.kryoserializer.buffer.max\", \"1024m\") \\\n    .config(\"spark.sql.shuffle.partitions\", \"500\") \\\n    .config(\"spark.sql.autoBroadcastJoinThreshold\", -1) \\\n    .config(\"spark.driver.memory\", \"8g\") \\\n    .config(\"spark.executor.memory\", \"8g\") \\\n    .getOrCreate()\n\nprint(spark.version)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:22:26.849423Z","iopub.execute_input":"2024-11-25T06:22:26.849923Z","iopub.status.idle":"2024-11-25T06:22:27.406201Z","shell.execute_reply.started":"2024-11-25T06:22:26.849876Z","shell.execute_reply":"2024-11-25T06:22:27.404972Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql import SparkSession\n\n# 创建 SparkSession 或获取现有的 SparkSession\nspark = SparkSession.builder.appName(\"YourAppName\").getOrCreate()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:28:28.448148Z","iopub.execute_input":"2024-11-25T06:28:28.448583Z","iopub.status.idle":"2024-11-25T06:28:28.457989Z","shell.execute_reply.started":"2024-11-25T06:28:28.448546Z","shell.execute_reply":"2024-11-25T06:28:28.456864Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql import SparkSession\n\nspark = SparkSession.builder.getOrCreate()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:28:35.647506Z","iopub.execute_input":"2024-11-25T06:28:35.647942Z","iopub.status.idle":"2024-11-25T06:28:35.655827Z","shell.execute_reply.started":"2024-11-25T06:28:35.647904Z","shell.execute_reply":"2024-11-25T06:28:35.654866Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import col\n\nals_data = als_data \\\n    .withColumn(\"user\", col(\"user\").cast(\"int\")) \\\n    .withColumn(\"item\", col(\"item\").cast(\"int\"))\n\nals_data.printSchema()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:28:37.738033Z","iopub.execute_input":"2024-11-25T06:28:37.738442Z","iopub.status.idle":"2024-11-25T06:28:37.762898Z","shell.execute_reply.started":"2024-11-25T06:28:37.738407Z","shell.execute_reply":"2024-11-25T06:28:37.761378Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n\n# 重新创建 SparkSession\nspark = SparkSession.builder \\\n    .appName(\"Recommendation System\") \\\n    .config(\"spark.executor.memory\", \"4g\") \\\n    .config(\"spark.driver.memory\", \"4g\") \\\n    .config(\"spark.kryoserializer.buffer.max\", \"1024m\") \\\n    .getOrCreate()\n\nprint(\"SparkSession restarted successfully.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.recommendation import ALS\nfrom pyspark.ml.evaluation import RegressionEvaluator","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:23:45.989994Z","iopub.execute_input":"2024-11-25T06:23:45.990384Z","iopub.status.idle":"2024-11-25T06:23:45.995716Z","shell.execute_reply.started":"2024-11-25T06:23:45.990354Z","shell.execute_reply":"2024-11-25T06:23:45.994494Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"als_data = transactions.select(col(\"customer_id\").alias(\"user\"), col(\"article_id\").alias(\"item\"), col(\"price\").alias(\"rating\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:23:48.052731Z","iopub.execute_input":"2024-11-25T06:23:48.053653Z","iopub.status.idle":"2024-11-25T06:23:48.075498Z","shell.execute_reply.started":"2024-11-25T06:23:48.053559Z","shell.execute_reply":"2024-11-25T06:23:48.074392Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"spark.conf.set(\"spark.sql.shuffle.partitions\", \"200\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:25:41.622363Z","iopub.execute_input":"2024-11-25T06:25:41.622773Z","iopub.status.idle":"2024-11-25T06:25:41.628488Z","shell.execute_reply.started":"2024-11-25T06:25:41.622739Z","shell.execute_reply":"2024-11-25T06:25:41.627181Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.feature import StringIndexer\n\n# 假设原始数据是 als_data，包含字符串类型的 user 和 item 列\nuser_indexer = StringIndexer(inputCol=\"user\", outputCol=\"user_index\")\nitem_indexer = StringIndexer(inputCol=\"item\", outputCol=\"item_index\")\n\n# 转换数据\nals_data = user_indexer.fit(als_data).transform(als_data)\nals_data = item_indexer.fit(als_data).transform(als_data)\n\n# 查看转换后的数据\nals_data.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:28:50.018001Z","iopub.execute_input":"2024-11-25T06:28:50.018413Z","iopub.status.idle":"2024-11-25T06:28:50.174033Z","shell.execute_reply.started":"2024-11-25T06:28:50.018377Z","shell.execute_reply":"2024-11-25T06:28:50.172066Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"als = ALS(userCol=\"user\", itemCol=\"item\", ratingCol=\"rating\", coldStartStrategy=\"drop\")\nmodel = als.fit(als_data)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-25T06:20:24.983195Z","iopub.execute_input":"2024-11-25T06:20:24.983625Z","iopub.status.idle":"2024-11-25T06:20:25.093515Z","shell.execute_reply.started":"2024-11-25T06:20:24.983585Z","shell.execute_reply":"2024-11-25T06:20:25.092046Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"6. 用上述推荐系统，向user_list各推荐3个产品，列出产品的名称和类别。","metadata":{}},{"cell_type":"code","source":"user_recs = model.recommendForUserSubset(als_data.filter(col(\"user\").isin(user_list)), 3)\nuser_recs.show(truncate=False)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"7. 请选择一组超参数，创建一个基于模型的服装推荐系统。（请以学号为随机种子分割训练集测试集）","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"8. 用上述推荐系统，向user_list各推荐3个产品，列出产品的名称和类别。","metadata":{}},{"cell_type":"code","source":"import os\n\nprint(os.listdir(\"/kaggle/input/h-and-m-personalized-fashion-recommendations\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-21T17:09:34.887103Z","iopub.execute_input":"2024-11-21T17:09:34.887602Z","iopub.status.idle":"2024-11-21T17:09:34.895337Z","shell.execute_reply.started":"2024-11-21T17:09:34.887561Z","shell.execute_reply":"2024-11-21T17:09:34.893702Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"9. 请使用交叉验证工具，测试最优的超参组合。最优模型的超参分别是什么？RMSE是多少？（请以学号为随机种子分割训练集测试集）","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"10. 用上述推荐系统，向user_list各推荐3个产品，列出产品的名称和类别。","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"11. 比较上述三种推荐模型的结果，你有什么看法？","metadata":{}},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null}]}