CA_NCF_tripadvisor.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "NCF Recommender with Explict Feedback"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Populating the interactive namespace from numpy and matplotlib\n"
     ]
    }
   ],
   "source": [
    "#Context-Aware Neural Collaborative Filtering (CA-NCF)\n",
    "#Author: Isam Mashhour Al Jawarneh\n",
    "#Date: 30 January 2019\n",
    "\n",
    "from zoo.pipeline.api.keras.layers import *\n",
    "from zoo.models.recommendation import *\n",
    "from zoo.models.recommendation.utils import *\n",
    "from zoo.common.nncontext import init_nncontext\n",
    "import os\n",
    "import sys\n",
    "import datetime as dt\n",
    "from bigdl.dataset.transformer import *\n",
    "from bigdl.dataset.base import *\n",
    "from bigdl.nn.criterion import *\n",
    "from bigdl.optim.optimizer import *\n",
    "from bigdl.optim import *\n",
    "from bigdl.util.common import *\n",
    "import matplotlib\n",
    "\n",
    "matplotlib.use('agg')\n",
    "import matplotlib.pyplot as plt\n",
    "%pylab inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = init_nncontext(create_spark_conf().setAppName(\"CA_NCF Tripadvisor\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy import stats\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "#import researchpy as rp\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#using pandas to read data from csv file\n",
    "\n",
    "df = pd.read_csv(\"datasets/tripAdvisor/Data_TripAdvisor_v2.csv\", names=['UserID', 'ItemID', 'rating','TripType'], header=None, usecols=[0,1,2,8])[['UserID', 'ItemID', 'rating','TripType']]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# for constructing objects containing items, the contexts and context conditions\n",
    "\n",
    "class MyClass(object):\n",
    "    def __init__(self, item,condition,context):\n",
    "        self.number = number\n",
    "        self.item = item\n",
    "        self.condition = condition\n",
    "        self.context = context"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/isam/.local/lib/python3.5/site-packages/numpy/core/fromnumeric.py:3506: RuntimeWarning: Degrees of freedom <= 0 for slice\n",
      "  **kwargs)\n",
      "/home/isam/.local/lib/python3.5/site-packages/numpy/core/_methods.py:209: RuntimeWarning: invalid value encountered in double_scalars\n",
      "  ret = ret.dtype.type(ret / rcount)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "item 503601 cmax is 4.225771273642583 con SOLO context TripType\n",
      "item 95716 cmax is 4.800568148193977 con SOLO context TripType\n",
      "item 88486 cmax is 4.024922359499622 con FAMILY context TripType\n",
      "item 91503 cmax is 4.024922359499622 con SOLO context TripType\n",
      "item 94389 cmax is inf con BUSINESS context TripType\n",
      "item 1382176 cmax is inf con FAMILY context TripType\n"
     ]
    }
   ],
   "source": [
    "######ITEM SPLITTING Procedure############\n",
    "## as described in :\n",
    "## Experimental evaluation of context-dependent collaborative filtering using item splitting\n",
    "## https://dl.acm.org/citation.cfm?id=2579997\n",
    "## output a list (item, condition, context), where condition is the context condition that will be used for splitting the item\n",
    "\n",
    "#profiling data\n",
    "# some_list will contain items marked to be split as they show t-statistics greater than a threshold (4.0 in this case)\n",
    "\n",
    "some_list = []\n",
    "# we have only one context factor in this dataset: TripType\n",
    "contextFactors = ['TripType']\n",
    "dflist = df['ItemID'].unique().tolist()\n",
    "cmax = 0.0\n",
    "con = ''\n",
    "context = ''\n",
    "for i in dflist:\n",
    "    cmax = 0.0\n",
    "    for j in contextFactors:\n",
    "        #print(j)\n",
    "        #retreiving unique context condistions of a spicific context factor (we have only one context factor herein)\n",
    "        conditionList = df[j].unique().tolist()\n",
    "        #array_length1 = len(conditionList)\n",
    "        #print(array_length1)\n",
    "        for k in conditionList:\n",
    "            #print(k)\n",
    "            #splitting an item into two items based on the current context condition\n",
    "            item1 = df[(df['ItemID'] == i) & (df[j] == k)]\n",
    "            item1.reset_index(inplace= True)  \n",
    "            item2 = df[(df['ItemID'] == i) & (df[j] != k)]\n",
    "            item2.reset_index(inplace= True)\n",
    "            l = stats.ttest_ind(item1['rating'], item2['rating'])\n",
    "            #if l.statistic > 4.0:\n",
    "            #print(i), print(l)\n",
    "            if (l.statistic > cmax) :\n",
    "                cmax = l.statistic\n",
    "                con = k\n",
    "                context = j\n",
    "            #print(cmax)\n",
    "        \n",
    "    #print('finished')\n",
    "    #HYPERPARAMETER cmax\n",
    "    #considering for splitting only items that show ttest greater than a threshold (4.0 in this case)\n",
    "    if cmax > 4.0:\n",
    "        print('item ' + str(i) + ' cmax is ' + str(cmax) + ' con ' + str(con) + ' context ' + str(context))\n",
    "        some_list.append(MyClass(str(i),str(con),str(context)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "#should be the same path in Jupyter where this notebook exists\n",
    "notebook_path = os.path.abspath(\"CA_NCF_tripadvisor.ipynb\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+------+---------+------------+----------+---------+------------+--------+\n",
      "|              UserID|ItemID|Rating|UserState|UserTimeZone|  ItemCity|ItemState|ItemTimeZone|TripType|\n",
      "+--------------------+------+------+---------+------------+----------+---------+------------+--------+\n",
      "|5C28F393B23BB8945...|219668|     5|       AK|          AK|GREENSBORO|       NC|     EASTERN|    SOLO|\n",
      "|3FA27F6E8AC712A82...|223860|     5|       AK|          AK|   PHOENIX|       AZ|    MOUNTAIN|    SOLO|\n",
      "|B99CFBB5411EDC888...| 75680|     5|       AK|          AK|   ANAHEIM|       CA|     PACIFIC|  FAMILY|\n",
      "|3FA27F6E8AC712A82...|224783|     5|       AK|          AK|   SEATTLE|       WA|     PACIFIC|    SOLO|\n",
      "|7CEFF5C32BA1F3B18...|222984|     5|       AK|          AK|     MIAMI|       MI|     EASTERN| COUPLES|\n",
      "+--------------------+------+------+---------+------------+----------+---------+------------+--------+\n",
      "only showing top 5 rows\n",
      "\n",
      "UserID :   2371\n",
      "ItemID :   2269\n"
     ]
    }
   ],
   "source": [
    "sqlContext = SQLContext(sc)\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql.functions import col, udf\n",
    "\n",
    "#reading raw data from CSV into a DataFrame\n",
    "\n",
    "Rating = Row(\"UserID\", \"ItemID\", \"Rating\",\"UserState\",\"UserTimeZone\",\"ItemCity\",\"ItemState\",\"ItemTimeZone\",\"TripType\")\n",
    "\n",
    "ratingsget = os.path.join(os.path.dirname(notebook_path), \"datasets/tripAdvisor/Data_TripAdvisor_v2.csv\")\n",
    "ratings= sc.textFile(ratingsget) \\\n",
    "    .map(lambda line: line.split(\",\")[0:9])\\\n",
    "    .map(lambda line: (line[0], int(line[1]), int(line[2]), line[3],line[4],line[5],line[6],line[7],line[8]))\\\n",
    "    .map(lambda r: Rating(*r))\n",
    "ratingDF_OLD_VERY = sqlContext.createDataFrame(ratings)\n",
    "ratingDF_OLD_VERY.show(5)\n",
    "\n",
    "someDFtt = ratingDF_OLD_VERY.select(col(\"UserID\")).distinct()\n",
    "\n",
    "#max_movie_id_alt = someDFt.count()\n",
    "\n",
    "print(\"UserID :   \" + str(someDFtt.count()))\n",
    "\n",
    "someDFtt1 = ratingDF_OLD_VERY.select(col(\"ItemID\")).distinct()\n",
    "\n",
    "#max_movie_id_alt = someDFt.count()\n",
    "\n",
    "print(\"ItemID :   \" + str(someDFtt1.count()))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "####all the following cells are preparations for incorporating context into original items\n",
    "####we are actual constructing first new artificial items set\n",
    "####then we apply the NCF form the library as-is"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "populateList = []\n",
    "def condition(i):\n",
    "    #note the permutation for each possible context condition, this will be used for binarization hereafter\n",
    "    switcher={\n",
    "            'COUPLES':[\"COUPLES\", \"SOLO\", \"FAMILY\", \"FRIENDS\" ,\"BUSINESS\"],\n",
    "            'SOLO':[\"SOLO\", \"COUPLES\", \"FAMILY\", \"FRIENDS\" ,\"BUSINESS\"],\n",
    "            'FAMILY':[\"FAMILY\", \"SOLO\", \"COUPLES\", \"FRIENDS\" ,\"BUSINESS\"],\n",
    "            'FRIENDS':[\"FRIENDS\", \"SOLO\", \"FAMILY\", \"COUPLES\" ,\"BUSINESS\"],\n",
    "            'BUSINESS':[\"BUSINESS\", \"SOLO\", \"FAMILY\", \"COUPLES\" ,\"FRIENDS\"],\n",
    "         \n",
    "           \n",
    "         }\n",
    "    return switcher.get(i,\"Invalid year\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, udf\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, OneHotEncoder, Binarizer\n",
    "from pyspark.sql.functions import rand \n",
    "#transformations\n",
    "indexer = StringIndexer(inputCol=\"UserID\", outputCol=\"newUserID1\")\n",
    "indexer1 = StringIndexer(inputCol=\"ItemID\", outputCol=\"indexedItemID2\")\n",
    "indexer2 = StringIndexer(inputCol=\"artItemID1\", outputCol=\"artItemID2\")\n",
    "binarizer22 = Binarizer(threshold=1.0, inputCol=\"locationIndexed\", outputCol=\"binarized_location\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "#transformations: applying indexers to ratingDF_OLD_VERY in a pipeline (one after the other)\n",
    "pipeline = Pipeline(stages=[indexer,indexer1])\n",
    "ratingDF_OLD = pipeline.fit(ratingDF_OLD_VERY).transform(ratingDF_OLD_VERY)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['SOLO', 'COUPLES', 'FAMILY', 'FRIENDS', 'BUSINESS']\n",
      "TripType\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1|ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|20A4EFC475E6F8B29...|    2230.0|503601|               0.0|    SOLO|         619.0|\n",
      "|C5FB378CCB9FB9FFF...|    1829.0|503601|               0.0|    SOLO|         619.0|\n",
      "|718748B9BAAEB7915...|      93.0|503601|               0.0|    SOLO|         619.0|\n",
      "|05D57A581E9DAE14F...|      17.0|503601|               1.0|BUSINESS|         619.0|\n",
      "|B95E2F08E14A7A45E...|    2220.0|503601|               1.0|BUSINESS|         619.0|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "['SOLO', 'COUPLES', 'FAMILY', 'FRIENDS', 'BUSINESS']\n",
      "TripType\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1|ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|80E85BC32CDB3F680...|     168.0| 95716|               0.0|    SOLO|         516.0|\n",
      "|2231F7619B1737F4B...|      54.0| 95716|               0.0|    SOLO|         516.0|\n",
      "|D7E558988172E3D3C...|      13.0| 95716|               0.0|    SOLO|         516.0|\n",
      "|1E76BA687E8C92E55...|     228.0| 95716|               0.0|    SOLO|         516.0|\n",
      "|1E76BA687E8C92E55...|     228.0| 95716|               0.0|    SOLO|         516.0|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "['FAMILY', 'SOLO', 'COUPLES', 'FRIENDS', 'BUSINESS']\n",
      "TripType\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1|ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|7DDBD21741629772D...|    2130.0| 88486|               0.0|  FAMILY|         788.0|\n",
      "|80E85BC32CDB3F680...|     168.0| 88486|               0.0|  FAMILY|         788.0|\n",
      "|5734024F8430E2343...|    1733.0| 88486|               0.0|  FAMILY|         788.0|\n",
      "|FBE33FB68A984F859...|     558.0| 88486|               1.0| COUPLES|         788.0|\n",
      "|8B189C66F4CA0A9DF...|    1947.0| 88486|               1.0| COUPLES|         788.0|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "\n",
      "['SOLO', 'COUPLES', 'FAMILY', 'FRIENDS', 'BUSINESS']\n",
      "TripType\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1|ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|FB7CA977608E02497...|     160.0| 91503|               0.0|    SOLO|         765.0|\n",
      "|8ECF93B4010544534...|    1173.0| 91503|               0.0|    SOLO|         765.0|\n",
      "|7BC0AF07CC240F2F6...|       1.0| 91503|               1.0|BUSINESS|         765.0|\n",
      "|AC050DB45BDCFF971...|      87.0| 91503|               1.0|BUSINESS|         765.0|\n",
      "|BE7F027FDB25A9D8D...|     470.0| 91503|               1.0|BUSINESS|         765.0|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "\n",
      "['BUSINESS', 'SOLO', 'FAMILY', 'COUPLES', 'FRIENDS']\n",
      "TripType\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1|ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "|751B8AB07EC80B6D7...|     318.0| 94389|               0.0|BUSINESS|         760.0|\n",
      "|BC04F877007456EC3...|     863.0| 94389|               0.0|BUSINESS|         760.0|\n",
      "|FBADE2112793ACB26...|    1766.0| 94389|               1.0|  FAMILY|         760.0|\n",
      "|0B3B44577ACCD2E91...|    2264.0| 94389|               1.0|  FAMILY|         760.0|\n",
      "|A8EBBC386D2AC8DF5...|     401.0| 94389|               1.0| COUPLES|         760.0|\n",
      "+--------------------+----------+------+------------------+--------+--------------+\n",
      "\n",
      "['FAMILY', 'SOLO', 'COUPLES', 'FRIENDS', 'BUSINESS']\n",
      "TripType\n",
      "+--------------------+----------+-------+------------------+--------+--------------+\n",
      "|              UserID|newUserID1| ItemID|binarized_location|TripType|indexedItemID2|\n",
      "+--------------------+----------+-------+------------------+--------+--------------+\n",
      "|4AB2E80C2EE2A335E...|    2195.0|1382176|               0.0|  FAMILY|         658.0|\n",
      "|7DAB6D856459A8A42...|     727.0|1382176|               0.0|  FAMILY|         658.0|\n",
      "|565CF711BCC62B264...|     447.0|1382176|               0.0|  FAMILY|         658.0|\n",
      "|41B1727D20FD42019...|     855.0|1382176|               0.0|  FAMILY|         658.0|\n",
      "|B6C7507B1C6334DA2...|    1287.0|1382176|               1.0| COUPLES|         658.0|\n",
      "+--------------------+----------+-------+------------------+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from functools import reduce\n",
    "from pyspark.sql import DataFrame\n",
    "from pyspark.sql.functions import col, udf\n",
    "df_list = []\n",
    "#preparing for the application of itemSplitting procedure above by separating items to-be-split from the original list\n",
    "toDiscard_df_list = []\n",
    "\n",
    "#remember that some_list contains split items\n",
    "for obj in some_list:\n",
    "    itemid= obj.item\n",
    "    ctxCond = obj.condition\n",
    "    ctx = obj.context\n",
    "    partialDF = ratingDF_OLD.select(\"*\").where((col(\"ItemID\")==itemid) )\n",
    "    \n",
    "    #toDiscard_df_list is a list of DFs, will be union later on\n",
    "    toDiscard_df_list.append(partialDF)\n",
    "    #print(partialDF.count())\n",
    "    # applying the switcher above to check based on what we split the items (binarization is the key here)\n",
    "    populateList = condition(ctxCond)\n",
    "    print(populateList)\n",
    "    print(ctx)\n",
    "    #a user defined function\n",
    "    general_udf = udf(lambda somename: categorical_from_vocab_list(somename, populateList, start=1))\n",
    "    partialDF_indexed = partialDF.withColumn(\"locationIndexed\", general_udf(col(ctx)).cast(\"double\"))\n",
    "    #print(partialDF_indexed.where(col('locationIndexed') != 0.0).count())\n",
    "    #print(partialDF_indexed.count())\n",
    "    pipeline1 = Pipeline(stages = [binarizer22])\n",
    "    allDF_temp =  pipeline1.fit(partialDF_indexed).transform(partialDF_indexed)\n",
    "    allDF_temp.select('UserID','newUserID1','ItemID','binarized_location','TripType','indexedItemID2').show(5)\n",
    "\n",
    "    bucket_cross_udf = udf(lambda feature1, feature2: hash_bucket(str(feature1) + \"_\" + str(feature2), bucket_size=100))\n",
    "    df_list.append(allDF_temp.withColumn(\"artItemID1\", bucket_cross_udf(col(\"indexedItemID2\"), col(\"binarized_location\")).cast(\"int\") + 1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "allDF_prepare = reduce(DataFrame.unionAll, df_list)#.cache()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+------+---------+------------+--------+---------+------------+--------+----------+--------------+---------------+------------------+----------+\n",
      "|              UserID|ItemID|Rating|UserState|UserTimeZone|ItemCity|ItemState|ItemTimeZone|TripType|newUserID1|indexedItemID2|locationIndexed|binarized_location|artItemID1|\n",
      "+--------------------+------+------+---------+------------+--------+---------+------------+--------+----------+--------------+---------------+------------------+----------+\n",
      "|20A4EFC475E6F8B29...|503601|     5|       CA|     PACIFIC|LASVEGAS|       NV|     PACIFIC|    SOLO|    2230.0|         619.0|            1.0|               0.0|        90|\n",
      "+--------------------+------+------+---------+------------+--------+---------+------------+--------+----------+--------------+---------------+------------------+----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "allDF_prepare.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import lit\n",
    "#discarding all partial DFs in the toDiscard_df_list constructed beforehand\n",
    "toDiscard_df_all = reduce(DataFrame.unionAll, toDiscard_df_list)#.cache()\n",
    "#print(toDiscard_df_all.select(col('artItemID1')).distinct().count())\n",
    "#toDiscard_df_all_sub now contains all items to split\n",
    "toDiscard_df_all_sub = toDiscard_df_all.select(\"*\")\n",
    "ratingDF_OLD_sub = ratingDF_OLD#.withColumn(\"indexedItemID2\",col(\"ItemID\"))\n",
    "#extract all items except split items\n",
    "remainingDF = ratingDF_OLD_sub.subtract(toDiscard_df_all_sub)\n",
    "\n",
    "remainingDF_indexed = remainingDF.withColumn('binarized_location',lit(1.0))\n",
    "bucket_cross_udf1 = udf(lambda feature1, feature2: hash_bucket(str(feature1) + \"_\" + str(feature2), bucket_size=1200000))\n",
    "remainingDF_indexed_all = remainingDF_indexed.withColumn(\"artItemID1\", bucket_cross_udf1(col(\"ItemID\"), col(\"binarized_location\")).cast(\"int\") + 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "#transformations\n",
    "remainingDF_indexed_all_sub = remainingDF_indexed_all.select(col('newUserID1') ,col('indexedItemID2'),col(\"UserID\"),\n",
    "                                        col('Rating'),col('ItemID'),col('artItemID1'),col(\"binarized_location\"))\n",
    "allDF_prepare_sub_sub = allDF_prepare.select(col(\"newUserID1\") ,col(\"indexedItemID2\"),col(\"UserID\"),\n",
    "                                        col(\"Rating\"),col(\"ItemID\"),col(\"artItemID1\"),col(\"binarized_location\"))\n",
    "allDF_difinite = remainingDF_indexed_all_sub.union(allDF_prepare_sub_sub)#.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "someDF_art = allDF_difinite.select(col(\"artItemID1\")).distinct()\n",
    "#print(someDF_art.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline2 = Pipeline(stages = [indexer2])\n",
    "\n",
    "allDF =  pipeline2.fit(allDF_difinite).transform(allDF_difinite)\n",
    "\n",
    "someDF1 = allDF.select((col(\"newUserID1\").cast(\"int\") + 1).alias(\"newUserID\"),col(\"indexedItemID2\").cast(\"int\"),\n",
    "                       col(\"Rating\"),col(\"ItemID\"),(col(\"artItemID2\").cast(\"int\") + 1).alias(\"aritid\"))\n",
    "someDF2 = allDF.select((col(\"newUserID1\").cast(\"int\") + 1).alias(\"newUserID\"),\n",
    "                       (col(\"artItemID2\").cast(\"int\") + 1).alias(\"artItemID\"),col(\"Rating\"))#.distinct()\n",
    "\n",
    "someDF_OLD = someDF2.orderBy(rand())#.select(\"*\").limit(5000000)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#  discard items with less than 20 ratings (interactions)\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark.sql import Window\n",
    "w = Window.partitionBy('artItemID')\n",
    "someDF_NEW = someDF_OLD.withColumn('count_artItemID', f.count('newUserID').over(w))\\\n",
    "    .where(f.col('count_artItemID') > 5)\\\n",
    "    .drop('count_artItemID')#.select(\"*\").limit(5000000)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "w = Window.partitionBy('newUserID')\n",
    "\n",
    "#the final dataFrame that contains the new list with items split\n",
    "someDF = someDF_NEW.withColumn('count_newUserID', f.count('artItemID').over(w))\\\n",
    ".where(f.col('count_newUserID') > 1).drop('count_newUserID')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "####now that we have constructed artificial items set, we are then ready to apply NCF from BigDL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "comprehensing your data. \n",
    "a tuple is (userid, movieid, rating). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "max newUserID 2371\n",
      "max artItemID 747\n",
      "2371 747\n"
     ]
    }
   ],
   "source": [
    "#we need this information to feed it to the NCF model in bigDL\n",
    "someDFt3 = someDF.select(col(\"newUserID\")).distinct()\n",
    "\n",
    "max_user_id1 =  someDF.agg({\"newUserID\": \"max\"}).collect()[0][0]\n",
    "print(\"max newUserID \" + str(someDF.agg({\"newUserID\": \"max\"}).collect()[0][0]))\n",
    "\n",
    "someDFt4 = someDF.select(col(\"artItemID\")).distinct()\n",
    "print(\"max artItemID \" + str(someDF.agg({\"artItemID\": \"max\"}).collect()[0][0]))\n",
    "max_item_id1 = someDF.agg({\"artItemID\": \"max\"}).collect()[0][0]\n",
    "print(max_user_id1, max_item_id1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Transforming raw rarings into an RDD . BigDl 6.0 requires input as RDD of samples, \n",
    "BigDL data structure that can be formed by two numpy arrays containing features and lables for the first and \n",
    "second array, respectively. \n",
    "in the following format Sample.from_ndarray(feature, label)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[<zoo.models.recommendation.recommender.UserItemFeature at 0x7f2c5620fa58>,\n",
       " <zoo.models.recommendation.recommender.UserItemFeature at 0x7f2c5620fb70>,\n",
       " <zoo.models.recommendation.recommender.UserItemFeature at 0x7f2c5620fb38>]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "def build_sample(user_id, item_id, rating):\n",
    "    sample = Sample.from_ndarray(np.array([user_id, item_id]), np.array([rating]))\n",
    "    return UserItemFeature(user_id, item_id, sample)\n",
    "#CA-NCF settings\n",
    "pairFeatureRdds_CANCF = someDF.rdd.map(lambda x: build_sample(x[0], x[1],x[2]))\n",
    "\n",
    "pairFeatureRdds_CANCF.repartition(1)\n",
    "\n",
    "pairFeatureRdds_CANCF.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "creating: createZooNeuralCF\n"
     ]
    }
   ],
   "source": [
    "#see more parameters here:\n",
    "#https://github.com/intel-analytics/analytics-zoo/blob/master/docs/docs/APIGuide/Models/recommendation.md\n",
    "#extra hyperparameters , user_embed=20, item_embed=20\n",
    "ncf = NeuralCF(user_count=max_user_id1, item_count=max_item_id1, class_num=5, hidden_layers=[40,20])#, include_mf = True,mf_embed = 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "millis = int(round(time.time() * 1000))\n",
    "##80% training data, 20% testing, randomly split\n",
    "#CA-NCF settings\n",
    "trainPairFeatureRdds_CANCF, valPairFeatureRdds_CANCF = pairFeatureRdds_CANCF.randomSplit([0.8, 0.2], seed = millis)\n",
    "valPairFeatureRdds_CANCF.cache()\n",
    "train_rdd_CANCF= trainPairFeatureRdds_CANCF.map(lambda pair_feature: pair_feature.sample)\n",
    "val_rdd_CANCF= valPairFeatureRdds_CANCF.map(lambda pair_feature: pair_feature.sample)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "UserItemPrediction [user_id: 148, item_id: 192, prediction: 5, probability: 0.20879336842252322]\n",
      "UserItemPrediction [user_id: 148, item_id: 422, prediction: 2, probability: 0.20849703814530976]\n",
      "UserItemPrediction [user_id: 833, item_id: 353, prediction: 5, probability: 0.20615691935021]\n",
      "UserItemPrediction [user_id: 1238, item_id: 83, prediction: 2, probability: 0.20535563770586102]\n",
      "UserItemPrediction [user_id: 1580, item_id: 578, prediction: 2, probability: 0.20632552923607234]\n"
     ]
    }
   ],
   "source": [
    "##predicting user-item pairs using CA-NCF\n",
    "predictUserItem_CANCF = ncf.predict_user_item_pair(valPairFeatureRdds_CANCF)\n",
    "for ans in predictUserItem_CANCF.take(5): print(ans)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "UserItemPrediction [user_id: 1200, item_id: 716, prediction: 5, probability: 0.2096270549772194]\n",
      "UserItemPrediction [user_id: 2000, item_id: 164, prediction: 5, probability: 0.2079028628742416]\n",
      "UserItemPrediction [user_id: 600, item_id: 162, prediction: 2, probability: 0.20755526144083172]\n",
      "UserItemPrediction [user_id: 600, item_id: 410, prediction: 2, probability: 0.20653570996804926]\n",
      "UserItemPrediction [user_id: 400, item_id: 462, prediction: 5, probability: 0.2071276076240727]\n",
      "UserItemPrediction [user_id: 400, item_id: 476, prediction: 5, probability: 0.20706403667906573]\n",
      "UserItemPrediction [user_id: 401, item_id: 287, prediction: 2, probability: 0.20675529799137107]\n",
      "UserItemPrediction [user_id: 201, item_id: 734, prediction: 5, probability: 0.20874777472144673]\n",
      "UserItemPrediction [user_id: 201, item_id: 337, prediction: 5, probability: 0.20851518292697024]\n",
      "UserItemPrediction [user_id: 201, item_id: 18, prediction: 5, probability: 0.20807441378078098]\n"
     ]
    }
   ],
   "source": [
    "#Recommending 5 items for every user\n",
    "recItemsForUsers_CANCF = ncf.recommend_for_user(valPairFeatureRdds_CANCF, 5)\n",
    "for ans in recItemsForUsers_CANCF.take(10): print(ans)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}