{"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":"We would first like to preface that this notebook is used to demonstrate both Pyspark and LSTM on kaggle, which is unfortunately not a distributed system and could not handle the training data in entirety. \n\nTherefore, in this notebook we are only using partial data, which we believe should be sufficient enough for a demonstration. Note that the results mentioned in the report uses the entire training data on a paid distributed service on AWS, which we cannot demonstrate here. ","metadata":{}},{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session\nshutil.rmtree(\"/kaggle/working\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2022-05-01T04:57:34.259974Z","iopub.execute_input":"2022-05-01T04:57:34.26056Z","iopub.status.idle":"2022-05-01T04:58:39.606953Z","shell.execute_reply.started":"2022-05-01T04:57:34.26045Z","shell.execute_reply":"2022-05-01T04:58:39.606145Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"code","source":"!pip install pyspark\n!pip install recbole","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:48:07.126201Z","iopub.execute_input":"2022-05-01T06:48:07.126536Z","iopub.status.idle":"2022-05-01T06:49:07.904305Z","shell.execute_reply.started":"2022-05-01T06:48:07.126449Z","shell.execute_reply":"2022-05-01T06:49:07.903461Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import glob\nimport os\nimport pyspark\n\nfrom pyspark.sql import SparkSession\n\nspark = SparkSession.builder \\\n     .master(\"local[*]\") \\\n     .config(\"spark.executor.memory\", \"70g\") \\\n     .config(\"spark.driver.memory\", \"50g\") \\\n     .config(\"spark.memory.offHeap.enabled\",True) \\\n     .config(\"spark.memory.offHeap.size\",\"16g\") \\\n     .appName(\"sampleCodeForReference\") \\\n     .getOrCreate()\n        \nspark = SparkSession(spark)","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:07.906297Z","iopub.execute_input":"2022-05-01T06:49:07.906559Z","iopub.status.idle":"2022-05-01T06:49:14.485920Z","shell.execute_reply.started":"2022-05-01T06:49:07.906525Z","shell.execute_reply":"2022-05-01T06:49:14.485034Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import numpy\n\nfrom pyspark.sql.functions import *\nfrom pyspark.sql.types import *\nfrom pyspark.sql.window import Window","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:14.487305Z","iopub.execute_input":"2022-05-01T06:49:14.487572Z","iopub.status.idle":"2022-05-01T06:49:14.505558Z","shell.execute_reply.started":"2022-05-01T06:49:14.487537Z","shell.execute_reply":"2022-05-01T06:49:14.504858Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"To save some time, we have already prepared the input inter file, just run the cell below","metadata":{}},{"cell_type":"code","source":"!cp -r ../input/pyspark-output /kaggle/working\n!mv /kaggle/working/pyspark-output /kaggle/working/recbox_data\n","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:14.509969Z","iopub.execute_input":"2022-05-01T06:49:14.511955Z","iopub.status.idle":"2022-05-01T06:49:20.565603Z","shell.execute_reply.started":"2022-05-01T06:49:14.511902Z","shell.execute_reply":"2022-05-01T06:49:20.564317Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Alternatively, you may run the following two cells to see how the inter file was obtained using Pyspark.","metadata":{}},{"cell_type":"code","source":"t_df = spark.read.option(\"header\",True) \\\n              .csv(\"../input/h-and-m-personalized-fashion-recommendations/transactions_train.csv\")\nt_df = t_df.withColumn(\"time_stamp\", t_df['t_dat'].cast(DateType()))\nt_df = t_df.withColumn(\"time_hash\", unix_timestamp(t_df[\"time_stamp\"])).select(col(\"customer_id\"), col(\"article_id\"), col(\"price\"), col(\"sales_channel_id\"), col(\"time_hash\"))\nt_df = t_df.select(col(\"customer_id\").alias(\"cid:token\"), col(\"article_id\").alias(\"aid:token\"), col(\"time_hash\").alias(\"t_hash:float\"))\nt_df = t_df.filter(col(\"t_hash:float\") > 1585620000)\nt_df.show()\n","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:45:29.499796Z","iopub.execute_input":"2022-05-01T06:45:29.500133Z","iopub.status.idle":"2022-05-01T06:46:01.328943Z","shell.execute_reply.started":"2022-05-01T06:45:29.500095Z","shell.execute_reply":"2022-05-01T06:46:01.328001Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!rm -r /kaggle/working/recbox_data\n!mkdir /kaggle/working/recbox_data\n\nimport pandas as pd\n\n#t_df.coalesce(1).write.option(\"header\", \"false\").csv('/kaggle/working/recbox_data/recbox_data.inter')\nt_df.toPandas().to_csv('/kaggle/working/recbox_data/recbox_data.inter', index=False, sep='\\t')","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:46:01.330261Z","iopub.execute_input":"2022-05-01T06:46:01.330585Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import logging\nfrom logging import getLogger\nfrom recbole.config import Config\nfrom recbole.data import create_dataset, data_preparation\nfrom recbole.model.sequential_recommender import GRU4Rec\nfrom recbole.trainer import Trainer\nfrom recbole.utils import init_seed, init_logger","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:47.453319Z","iopub.execute_input":"2022-05-01T06:49:47.453589Z","iopub.status.idle":"2022-05-01T06:49:49.704393Z","shell.execute_reply.started":"2022-05-01T06:49:47.453560Z","shell.execute_reply":"2022-05-01T06:49:49.703634Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"parameters = {\n    'user_inter_num_interval': \"[30,inf)\",\n    'item_inter_num_interval': \"[40,inf)\",\n    'epochs': 50,\n    'data_path': '/kaggle/working',\n    'USER_ID_FIELD': 'cid',\n    'ITEM_ID_FIELD': 'aid',\n    'TIME_FIELD': 't_hash',\n    'load_col': {'inter': ['cid', 'aid', 't_hash']},\n    'neg_sampling': None,\n    'eval_args': {\n        'split': {'RS': [9, 0, 1]},\n        'group_by': 'user',\n        'order': 'TO',\n        'mode': 'full'}\n}\n","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:49.706239Z","iopub.execute_input":"2022-05-01T06:49:49.706539Z","iopub.status.idle":"2022-05-01T06:49:49.712505Z","shell.execute_reply.started":"2022-05-01T06:49:49.706503Z","shell.execute_reply":"2022-05-01T06:49:49.711823Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"new_config = Config(model='GRU4Rec', dataset='recbox_data', config_dict=parameters)\ninit_seed(new_config['seed'], new_config['reproducibility'])\ninit_logger(new_config)\nlog = getLogger()\nstream_handler = logging.StreamHandler()\nstream_handler.setLevel(logging.INFO)\nlog.addHandler(stream_handler)\nlog.info(new_config)","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:50.608716Z","iopub.execute_input":"2022-05-01T06:49:50.609339Z","iopub.status.idle":"2022-05-01T06:49:50.933457Z","shell.execute_reply.started":"2022-05-01T06:49:50.609299Z","shell.execute_reply":"2022-05-01T06:49:50.932858Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"dataset = create_dataset(new_config)\nlog.info(dataset)","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:49:55.854911Z","iopub.execute_input":"2022-05-01T06:49:55.855184Z","iopub.status.idle":"2022-05-01T06:51:12.021583Z","shell.execute_reply.started":"2022-05-01T06:49:55.855155Z","shell.execute_reply":"2022-05-01T06:51:12.020917Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!rm -r /kaggle/working/saved","metadata":{"execution":{"iopub.status.busy":"2022-04-20T17:13:26.404089Z","iopub.execute_input":"2022-04-20T17:13:26.404854Z","iopub.status.idle":"2022-04-20T17:13:27.089938Z","shell.execute_reply.started":"2022-04-20T17:13:26.404813Z","shell.execute_reply":"2022-04-20T17:13:27.089069Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train, valid, test = data_preparation(new_config, dataset)\nmodel = GRU4Rec(new_config, train.dataset).to(new_config['device'])\nlog.info(model)\ntrainer = Trainer(new_config, model)\nbest_valid_score, best_valid_result = trainer.fit(train)","metadata":{"execution":{"iopub.status.busy":"2022-05-01T06:51:31.190659Z","iopub.execute_input":"2022-05-01T06:51:31.191601Z","iopub.status.idle":"2022-05-01T07:10:16.154185Z","shell.execute_reply.started":"2022-05-01T06:51:31.191551Z","shell.execute_reply":"2022-05-01T07:10:16.153202Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from recbole.utils.case_study import full_sort_topk\n\nex_users = dataset.id2token(dataset.uid_field, list(range(dataset.user_num)))[1:]","metadata":{"execution":{"iopub.status.busy":"2022-05-01T07:11:33.840282Z","iopub.execute_input":"2022-05-01T07:11:33.840904Z","iopub.status.idle":"2022-05-01T07:11:33.856544Z","shell.execute_reply.started":"2022-05-01T07:11:33.840866Z","shell.execute_reply":"2022-05-01T07:11:33.855831Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"topk_items = []\nfor in_users in list(range(dataset.user_num))[1:]:\n    _, topk_iid_list = full_sort_topk([in_users], model, test, k=12, device=new_config['device'])\n    last_topk_iid_list = topk_iid_list[-1]\n    ext_items = dataset.id2token(dataset.iid_field, last_topk_iid_list.cpu()).tolist()\n    topk_items.append(ext_items)\nprint(len(topk_items))","metadata":{"execution":{"iopub.status.busy":"2022-05-01T07:11:37.762519Z","iopub.execute_input":"2022-05-01T07:11:37.762780Z","iopub.status.idle":"2022-05-01T07:12:50.875956Z","shell.execute_reply.started":"2022-05-01T07:11:37.762751Z","shell.execute_reply":"2022-05-01T07:12:50.875248Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"The cell below shows the output of the model","metadata":{}},{"cell_type":"code","source":"import pandas as pd\n\nexternal_item_str = [' '.join(x) for x in topk_items]\nresult = pd.DataFrame(ex_users, columns=['customer_id'])\nresult['prediction'] = external_item_str\nresult_df = spark.sparkContext.createDataFrame(result)\nresult_df.show()\n","metadata":{"execution":{"iopub.status.busy":"2022-05-01T07:14:20.886123Z","iopub.execute_input":"2022-05-01T07:14:20.886811Z","iopub.status.idle":"2022-05-01T07:14:27.357737Z","shell.execute_reply.started":"2022-05-01T07:14:20.886771Z","shell.execute_reply":"2022-05-01T07:14:27.356909Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"","metadata":{}}]}