{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"markdown","source":"This notebook explores several approaches to tackle the H&M recommendation problem with the usage of PySpark.\nDuring the work we have tried the following strategies:\n* Recommend previously purchased items\n* Recommend items that are frequently bought together\n* Recommend most popular items\n* Discuss the Spark's module for collaborative filtering: ALS","metadata":{}},{"cell_type":"markdown","source":"# Read the data","metadata":{}},{"cell_type":"code","source":"!pip install pyspark -q","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2022-04-09T04:40:10.256279Z","iopub.execute_input":"2022-04-09T04:40:10.256728Z","iopub.status.idle":"2022-04-09T04:41:09.235418Z","shell.execute_reply.started":"2022-04-09T04:40:10.256698Z","shell.execute_reply":"2022-04-09T04:41:09.23448Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import pyspark\nfrom pyspark.sql import SparkSession\n\nimport warnings\nwarnings.filterwarnings(\"ignore\")\n\nspark = SparkSession.builder.appName(\"H&M\").getOrCreate()","metadata":{"execution":{"iopub.status.busy":"2022-04-09T05:36:55.643779Z","iopub.execute_input":"2022-04-09T05:36:55.644133Z","iopub.status.idle":"2022-04-09T05:36:55.650234Z","shell.execute_reply.started":"2022-04-09T05:36:55.6441Z","shell.execute_reply":"2022-04-09T05:36:55.649319Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"transactions = (spark.read.format(\"csv\")\n .option(\"header\", \"true\")\n .load(\"../input/h-and-m-personalized-fashion-recommendations/transactions_train.csv\"))\ntransactions.printSchema()","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:41:15.473895Z","iopub.execute_input":"2022-04-09T04:41:15.474237Z","iopub.status.idle":"2022-04-09T04:41:21.099062Z","shell.execute_reply.started":"2022-04-09T04:41:15.474186Z","shell.execute_reply":"2022-04-09T04:41:21.097891Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.sql.functions import *\n\ntransactions = transactions.withColumn('t_dat', to_date(\"t_dat\"))","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:42:33.288781Z","iopub.execute_input":"2022-04-09T04:42:33.289593Z","iopub.status.idle":"2022-04-09T04:42:33.326004Z","shell.execute_reply.started":"2022-04-09T04:42:33.289541Z","shell.execute_reply":"2022-04-09T04:42:33.325194Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Analysis","metadata":{}},{"cell_type":"markdown","source":"Keep only last week purchases for each customer
","metadata":{}},{"cell_type":"code","source":"# find the latests purchase day for every customer\ntmp = transactions.groupby(\"customer_id\").agg(\n expr(\"max(t_dat) AS latest_date\")\n)\n\njoinExpression = transactions['customer_id'] == tmp[\"customer_id\"]\n\ntransactions = transactions.join(tmp, joinExpression).drop(tmp[\"customer_id\"])","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:29:53.470386Z","iopub.execute_input":"2022-04-08T17:29:53.471246Z","iopub.status.idle":"2022-04-08T17:29:53.707145Z","shell.execute_reply.started":"2022-04-08T17:29:53.471189Z","shell.execute_reply":"2022-04-08T17:29:53.706126Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"transactions = transactions.withColumn('date_diff', \n datediff(col(\"latest_date\"), col(\"t_dat\"))\n ).filter(\"date_diff <= 6\")","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:29:53.708503Z","iopub.execute_input":"2022-04-08T17:29:53.708817Z","iopub.status.idle":"2022-04-08T17:29:53.779137Z","shell.execute_reply.started":"2022-04-08T17:29:53.708773Z","shell.execute_reply":"2022-04-08T17:29:53.778198Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Recommend the items that are most popular to be purchased together
","metadata":{}},{"cell_type":"code","source":"tmp = (transactions.groupby(\"customer_id\", \"article_id\")\n .count())\ntmp.orderBy(\"count\", ascending=False).show(5)","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:29:53.780745Z","iopub.execute_input":"2022-04-08T17:29:53.781056Z","iopub.status.idle":"2022-04-08T17:31:52.52175Z","shell.execute_reply.started":"2022-04-08T17:29:53.780993Z","shell.execute_reply":"2022-04-08T17:31:52.520622Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"tmp.cache()\ntmp.createOrReplaceTempView(\"tmp\")\npaired_items = spark.sql(\"\"\"\n SELECT * FROM\n (SELECT *, row_number() over\n (PARTITION BY customer_id ORDER BY count DESC) as row_index\n FROM tmp) a\n WHERE row_index <= 3\n \"\"\")\n\ntmp.unpersist()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"paired_items = (paired_items.groupBy('customer_id')\n .agg(collect_set('article_id')\n .alias('article_id')))\npaired_items.show()","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:31:52.860976Z","iopub.execute_input":"2022-04-08T17:31:52.861232Z","iopub.status.idle":"2022-04-08T17:33:11.767783Z","shell.execute_reply.started":"2022-04-08T17:31:52.861202Z","shell.execute_reply":"2022-04-08T17:33:11.765053Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Recommend last week's most popular items
","metadata":{}},{"cell_type":"code","source":"transactions.cache()\ntransactions.createOrReplaceTempView('transactions')\ntop12 = spark.sql(\"\"\"\n SELECT article_id, COUNT(*) count FROM transactions\n WHERE t_dat > '2020-08-23'\n Group BY article_id\n ORDER by count DESC\n Limit 12\n \"\"\")","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:34:40.51728Z","iopub.execute_input":"2022-04-08T17:34:40.517604Z","iopub.status.idle":"2022-04-08T17:34:40.578806Z","shell.execute_reply.started":"2022-04-08T17:34:40.51756Z","shell.execute_reply":"2022-04-08T17:34:40.578106Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.sql import Row\nimport pandas as pd\n\np_top12 = top12.select('article_id').toPandas()\nt12 = ' '.join(p[0] for p in p_top12.to_numpy())\nprint(t12)","metadata":{"execution":{"iopub.status.busy":"2022-04-08T17:34:40.579766Z","iopub.execute_input":"2022-04-08T17:34:40.579988Z","iopub.status.idle":"2022-04-08T17:35:15.158975Z","shell.execute_reply.started":"2022-04-08T17:34:40.579962Z","shell.execute_reply":"2022-04-08T17:35:15.158053Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# ALS","metadata":{}},{"cell_type":"markdown","source":"Goal: Factorize the given ratings matrix $R$ into two factors-user matrix $U$ and item matrix $V$ such that $R \\approx U^TV$.\n\nNotation: \n* $u_i$-the $i$th column of the user matrix\n* $v_i$-the $i$th column of the item matrix\n* $r_{ij}$-the rating of the $j$th item from $i$th user\n* \\lambda-regularization factor\n* $n_{u_i}$-the number of items the $i$th user rated\n* $n_{v_j}$-the number of time the $j$th item was rated\n\nObjective:\n$\\text{argmin}_{U,V}\\sum_{i,j, r_{ij}\\not=0}(r_{ij}-u_i^Tv_j)^2+\\lambda(\\sum_{i}n_{u_i}\\|u_i\\|^2+\\sum_{j}n_{v_j}\\|v_j\\|^2)$\n\nAlgorithm: Fix the set of variables $U$ and treat them as constants, the objective is convex function of $V$. Solve for $V$. Repeat similarly for $U$.\n\nALS finds $k$-dimensional feature vector for each user and item such that the dot product of their feature vectors approximates the user's rating for the item.\n\nALS requires an input dataset with only three columns: a user ID column, an item ID column, and a rating column. Moreover, ALS handles both explicit ratings-a numerical rating and the implicit ratings-the strength of interactions between a user and the item (here the number of purchases for the given user of the given item).","metadata":{}},{"cell_type":"code","source":"from pyspark.ml.evaluation import RegressionEvaluator\nfrom pyspark.ml.recommendation import ALS","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:42:18.539474Z","iopub.execute_input":"2022-04-09T04:42:18.539757Z","iopub.status.idle":"2022-04-09T04:42:18.841908Z","shell.execute_reply.started":"2022-04-09T04:42:18.539729Z","shell.execute_reply":"2022-04-09T04:42:18.841149Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"tmp = transactions.withColumn('t_dat', transactions['t_dat'].cast('string'))\ntmp = tmp.withColumn('date', from_unixtime(unix_timestamp('t_dat', 'yyyy-MM-dd')))\ntmp = tmp.withColumn('year', year(col('date')))\ntmp = tmp.withColumn('month', month(col('date')))\ntmp = tmp.withColumn('day', date_format(col('date'), \"d\"))\n\ntmp = tmp[tmp['year'] == 2020]\ntmp = tmp[tmp['month'] == 9]\ntmp = tmp[tmp['day'] == 22]\ntransactions.unpersist()\n\n# Prepare the dataset\ntmp = tmp.groupby('customer_id', 'article_id').count()\ntmp.show(5)","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:42:38.967401Z","iopub.execute_input":"2022-04-09T04:42:38.967984Z","iopub.status.idle":"2022-04-09T04:43:56.90908Z","shell.execute_reply.started":"2022-04-09T04:42:38.967944Z","shell.execute_reply":"2022-04-09T04:43:56.908191Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.ml.feature import StringIndexer\nfrom pyspark.ml import Pipeline\n\nindexer = [StringIndexer(inputCol=column, outputCol=column+\"_index\") \n for column in list(set(tmp.columns) - set(['count']))]","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:43:56.910662Z","iopub.execute_input":"2022-04-09T04:43:56.910957Z","iopub.status.idle":"2022-04-09T04:43:56.973682Z","shell.execute_reply.started":"2022-04-09T04:43:56.910918Z","shell.execute_reply":"2022-04-09T04:43:56.972958Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import gc\ngc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:43:56.974808Z","iopub.execute_input":"2022-04-09T04:43:56.975133Z","iopub.status.idle":"2022-04-09T04:43:57.098873Z","shell.execute_reply.started":"2022-04-09T04:43:56.975102Z","shell.execute_reply":"2022-04-09T04:43:57.098Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"pipeline = Pipeline(stages=indexer)\ntransformed = pipeline.fit(tmp).transform(tmp)\n\n(train, test) = transformed.randomSplit([0.8, 0.2])","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:43:57.100932Z","iopub.execute_input":"2022-04-09T04:43:57.101374Z","iopub.status.idle":"2022-04-09T04:46:27.861594Z","shell.execute_reply.started":"2022-04-09T04:43:57.101329Z","shell.execute_reply":"2022-04-09T04:46:27.860541Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"als = ALS(maxIter=5, regParam=0.09, \n rank=25, userCol=\"customer_id_index\",\n itemCol=\"article_id_index\", ratingCol=\"count\",\n coldStartStrategy=\"drop\", nonnegative=True)\n\nmodel=als.fit(train)","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:46:27.862965Z","iopub.execute_input":"2022-04-09T04:46:27.863757Z","iopub.status.idle":"2022-04-09T04:49:09.717092Z","shell.execute_reply.started":"2022-04-09T04:46:27.863711Z","shell.execute_reply":"2022-04-09T04:49:09.71613Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pyspark.ml.evaluation import RegressionEvaluator\n\nevaluator=RegressionEvaluator(metricName=\"rmse\",labelCol=\"count\",predictionCol=\"prediction\")\npredictions=model.transform(test)\nrmse=evaluator.evaluate(predictions)\nprint(\"RMSE is equal to : \", str(rmse))","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:49:09.721259Z","iopub.execute_input":"2022-04-09T04:49:09.721596Z","iopub.status.idle":"2022-04-09T04:50:24.576177Z","shell.execute_reply.started":"2022-04-09T04:49:09.72155Z","shell.execute_reply":"2022-04-09T04:50:24.574586Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":" model.recommendForAllUsers(10).show(10)","metadata":{"execution":{"iopub.status.busy":"2022-04-09T04:51:45.365378Z","iopub.execute_input":"2022-04-09T04:51:45.36688Z","iopub.status.idle":"2022-04-09T04:51:57.177231Z","shell.execute_reply.started":"2022-04-09T04:51:45.3668Z","shell.execute_reply":"2022-04-09T04:51:57.17596Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Since recommendations are given as indexes with should map them to ids.","metadata":{}},{"cell_type":"code","source":"import pandas as pd\n\nrecs = model.recommendForAllUsers(10).toPandas()\nrecommendations = (recs['recommendations'].apply(pd.Series)\n .merge(recs, right_index = True, left_index = True) # split a Pandas column with lists of tuples into separate columns\n .drop([\"recommendations\"], axis = 1) # drop the list of tuples\n .melt(id_vars = ['customer_id_index'], value_name = \"recommendations\") # turn 'recommendation' columns into rows\n .drop(\"variable\", axis = 1)\n .sort_values('customer_id_index')\n .dropna())\n\nrecommendations = pd.concat([recommendations['recommendations'].apply(pd.Series), \n recommendations['customer_id_index']], axis = 1) # separate article_ids from counts","metadata":{"execution":{"iopub.status.busy":"2022-04-09T05:27:36.569709Z","iopub.execute_input":"2022-04-09T05:27:36.570564Z","iopub.status.idle":"2022-04-09T05:28:00.063198Z","shell.execute_reply.started":"2022-04-09T05:27:36.570517Z","shell.execute_reply":"2022-04-09T05:28:00.062228Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"recommendations.columns = ['ArticleID_index','count','UserID_index']\ntransformed_subset = transformed.select('article_id', 'article_id_index', 'customer_id', 'customer_id_index')\ntransformed_subset = transformed_subset.toPandas()","metadata":{"execution":{"iopub.status.busy":"2022-04-09T05:28:11.227271Z","iopub.execute_input":"2022-04-09T05:28:11.22779Z","iopub.status.idle":"2022-04-09T05:29:41.814551Z","shell.execute_reply.started":"2022-04-09T05:28:11.227754Z","shell.execute_reply":"2022-04-09T05:29:41.813786Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# map index to id\narticle_map = dict(zip(transformed_subset['article_id_index'], transformed_subset['article_id']))\ncustomer_map = dict(zip(transformed_subset['customer_id_index'], transformed_subset['customer_id']))\nrecommendations['article_id'] = recommendations['ArticleID_index'].map(article_map)\nrecommendations['customer_id'] = recommendations['UserID_index'].map(customer_map)","metadata":{"execution":{"iopub.status.busy":"2022-04-09T05:30:39.374333Z","iopub.execute_input":"2022-04-09T05:30:39.374632Z","iopub.status.idle":"2022-04-09T05:30:39.433942Z","shell.execute_reply.started":"2022-04-09T05:30:39.374602Z","shell.execute_reply":"2022-04-09T05:30:39.433047Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"recommendations.reset_index(drop=True, inplace=True)\nrecommendations = recommendations[['customer_id','article_id']]\nrecommendations","metadata":{"execution":{"iopub.status.busy":"2022-04-09T05:36:04.134168Z","iopub.execute_input":"2022-04-09T05:36:04.135115Z","iopub.status.idle":"2022-04-09T05:36:04.154575Z","shell.execute_reply.started":"2022-04-09T05:36:04.135063Z","shell.execute_reply":"2022-04-09T05:36:04.153929Z"},"trusted":true},"execution_count":null,"outputs":[]}]}