Collaborative_Filtering (1).ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "kernelspec": {
      "display_name": "3point6",
      "language": "python",
      "name": "build_central"
    },
    "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.6.10"
    },
    "colab": {
      "name": "Collaborative Filtering.ipynb",
      "provenance": []
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "F1OqjnLnXq_Y"
      },
      "source": [
        "import numpy as np\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as pt\n",
        "from pyspark.ml import pipeline\n",
        "from pyspark.ml.recommendation import ALS, ALSModel\n",
        "from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder, CrossValidator\n",
        "from pyspark.ml.evaluation import RegressionEvaluator\n",
        "from pyspark.ml import Pipeline\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark.sql.types import *\n",
        "from pyspark.sql import Row\n",
        "from pyspark.sql.types import *"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5YXVzNnJXq_a"
      },
      "source": [
        "import findspark\n",
        "findspark.init()\n",
        "findspark.find()\n",
        "import itertools\n",
        "import pyspark\n",
        "import sys\n",
        "import time\n",
        "import json\n",
        "from pyspark import SparkContext, SparkConf\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql import SQLContext\n",
        "from pyspark.sql.functions  import date_format\n",
        "\n",
        "\n",
        "spark = SparkSession \\\n",
        "    .builder \\\n",
        "    .appName(\"Content Module\") \\\n",
        "    .getOrCreate()\n",
        "\n",
        "sqlContext = SQLContext(spark)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "UsBPPFBXXq_b",
        "outputId": "106942d0-a4d4-4bde-f05b-bd99fb99b2aa"
      },
      "source": [
        "business_df=spark.read.json('business.json')\n",
        "user_df=spark.read.json('user.json')\n",
        "review_df=spark.read.json('train_review (1).json')\n",
        "\n",
        "print('The schema of the business json file is as follows')\n",
        "business_df.printSchema()\n",
        "print('\\n\\n')\n",
        "print('The schema of the user json file is as follows')\n",
        "user_df.printSchema()\n",
        "print('\\n\\n')\n",
        "print('The schema of the review json file is as follows')\n",
        "review_df.printSchema()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "The schema of the business json file is as follows\n",
            "root\n",
            " |-- address: string (nullable = true)\n",
            " |-- attributes: struct (nullable = true)\n",
            " |    |-- AcceptsInsurance: string (nullable = true)\n",
            " |    |-- AgesAllowed: string (nullable = true)\n",
            " |    |-- Alcohol: string (nullable = true)\n",
            " |    |-- Ambience: string (nullable = true)\n",
            " |    |-- BYOB: string (nullable = true)\n",
            " |    |-- BYOBCorkage: string (nullable = true)\n",
            " |    |-- BestNights: string (nullable = true)\n",
            " |    |-- BikeParking: string (nullable = true)\n",
            " |    |-- BusinessAcceptsBitcoin: string (nullable = true)\n",
            " |    |-- BusinessAcceptsCreditCards: string (nullable = true)\n",
            " |    |-- BusinessParking: string (nullable = true)\n",
            " |    |-- ByAppointmentOnly: string (nullable = true)\n",
            " |    |-- Caters: string (nullable = true)\n",
            " |    |-- CoatCheck: string (nullable = true)\n",
            " |    |-- Corkage: string (nullable = true)\n",
            " |    |-- DietaryRestrictions: string (nullable = true)\n",
            " |    |-- DogsAllowed: string (nullable = true)\n",
            " |    |-- DriveThru: string (nullable = true)\n",
            " |    |-- GoodForDancing: string (nullable = true)\n",
            " |    |-- GoodForKids: string (nullable = true)\n",
            " |    |-- GoodForMeal: string (nullable = true)\n",
            " |    |-- HairSpecializesIn: string (nullable = true)\n",
            " |    |-- HappyHour: string (nullable = true)\n",
            " |    |-- HasTV: string (nullable = true)\n",
            " |    |-- Music: string (nullable = true)\n",
            " |    |-- NoiseLevel: string (nullable = true)\n",
            " |    |-- Open24Hours: string (nullable = true)\n",
            " |    |-- OutdoorSeating: string (nullable = true)\n",
            " |    |-- RestaurantsAttire: string (nullable = true)\n",
            " |    |-- RestaurantsCounterService: string (nullable = true)\n",
            " |    |-- RestaurantsDelivery: string (nullable = true)\n",
            " |    |-- RestaurantsGoodForGroups: string (nullable = true)\n",
            " |    |-- RestaurantsPriceRange2: string (nullable = true)\n",
            " |    |-- RestaurantsReservations: string (nullable = true)\n",
            " |    |-- RestaurantsTableService: string (nullable = true)\n",
            " |    |-- RestaurantsTakeOut: string (nullable = true)\n",
            " |    |-- Smoking: string (nullable = true)\n",
            " |    |-- WheelchairAccessible: string (nullable = true)\n",
            " |    |-- WiFi: string (nullable = true)\n",
            " |-- business_id: string (nullable = true)\n",
            " |-- categories: string (nullable = true)\n",
            " |-- city: string (nullable = true)\n",
            " |-- hours: struct (nullable = true)\n",
            " |    |-- Friday: string (nullable = true)\n",
            " |    |-- Monday: string (nullable = true)\n",
            " |    |-- Saturday: string (nullable = true)\n",
            " |    |-- Sunday: string (nullable = true)\n",
            " |    |-- Thursday: string (nullable = true)\n",
            " |    |-- Tuesday: string (nullable = true)\n",
            " |    |-- Wednesday: string (nullable = true)\n",
            " |-- is_open: long (nullable = true)\n",
            " |-- latitude: double (nullable = true)\n",
            " |-- longitude: double (nullable = true)\n",
            " |-- name: string (nullable = true)\n",
            " |-- postal_code: string (nullable = true)\n",
            " |-- state: string (nullable = true)\n",
            "\n",
            "\n",
            "\n",
            "\n",
            "The schema of the user json file is as follows\n",
            "root\n",
            " |-- compliment_cool: long (nullable = true)\n",
            " |-- compliment_cute: long (nullable = true)\n",
            " |-- compliment_funny: long (nullable = true)\n",
            " |-- compliment_hot: long (nullable = true)\n",
            " |-- compliment_list: long (nullable = true)\n",
            " |-- compliment_more: long (nullable = true)\n",
            " |-- compliment_note: long (nullable = true)\n",
            " |-- compliment_photos: long (nullable = true)\n",
            " |-- compliment_plain: long (nullable = true)\n",
            " |-- compliment_profile: long (nullable = true)\n",
            " |-- compliment_writer: long (nullable = true)\n",
            " |-- cool: long (nullable = true)\n",
            " |-- elite: string (nullable = true)\n",
            " |-- fans: long (nullable = true)\n",
            " |-- friends: string (nullable = true)\n",
            " |-- funny: long (nullable = true)\n",
            " |-- name: string (nullable = true)\n",
            " |-- useful: long (nullable = true)\n",
            " |-- user_id: string (nullable = true)\n",
            " |-- yelping_since: string (nullable = true)\n",
            "\n",
            "\n",
            "\n",
            "\n",
            "The schema of the review json file is as follows\n",
            "root\n",
            " |-- business_id: string (nullable = true)\n",
            " |-- date: string (nullable = true)\n",
            " |-- review_id: string (nullable = true)\n",
            " |-- stars: double (nullable = true)\n",
            " |-- text: string (nullable = true)\n",
            " |-- user_id: string (nullable = true)\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fqLj5d5TXq_c"
      },
      "source": [
        "### Spark ALS implementation requires the rating matrix to have the follwoing data types:\n",
        "* ### ratings_df_schema = StructType(\n",
        "* ### [StructField('userId', IntegerType()),\n",
        "* ###  StructField('businessId', IntegerType()),\n",
        "* ### StructField('rating', DoubleType())])"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0noY6j5MXq_c"
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "n58XUp-UXq_d"
      },
      "source": [
        "business_df.select('business_id').rdd.map(lambda x:x[0]).zipWithIndex().take(5)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jQEx0nlMXq_d"
      },
      "source": [
        "b_idDict=business_df.select('business_id').rdd.map(lambda x:x[0]).zipWithIndex()\n",
        "b_idDataFrame=sqlContext.createDataFrame(b_idDict,StructType([StructField(\"business_id\", StringType(), True),StructField(\"businessId\", IntegerType(), True)]))\n",
        "b_idDataFrame.show(5)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "OSi-OlkMXq_e"
      },
      "source": [
        "b_idDataFrame.count()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "em3h52D9Xq_e"
      },
      "source": [
        "u_idDict=user_df.select('user_id').rdd.map(lambda x:x[0]).zipWithIndex()\n",
        "u_idDataFrame=sqlContext.createDataFrame(b_idDict,StructType([StructField(\"user_id\", StringType(), True),StructField(\"userId\", IntegerType(), True)]))\n",
        "u_idDataFrame.show(5)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "r3p8xuubXq_f"
      },
      "source": [
        "a = user_df.alias(\"a\")\n",
        "b = u_idDataFrame.alias(\"b\")\n",
        "    \n",
        "user_new = a.join(b, col(\"a.user_id\") == col(\"b.user_id\"), 'inner') \\\n",
        "             .select([col('a.'+xx) for xx in a.columns] + [col('b.userId')])\n",
        "\n",
        "user_new.select('userId','user_id', 'name').show(5,truncate=False)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Fd_6XwtkXq_f"
      },
      "source": [
        "# map new userId and businessId in the review dataframe\n",
        "\n",
        "review_df = review_df.select('user_id', 'business_id', 'stars')\n",
        "\n",
        "\n",
        "# map the userId\n",
        "a = review_df.alias(\"a\")\n",
        "b = user_new.alias(\"b\")\n",
        "    \n",
        "review1_df = a.join(b, col(\"a.user_id\") == col(\"b.user_id\"), 'inner') \\\n",
        "                     .select([col('a.'+xx) for xx in a.columns] + [col('b.userId')])\n",
        "\n",
        "review1_df.show(5)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yEL5HBn7Xq_g"
      },
      "source": [
        "# map the businessId\n",
        "a = review1_df.alias(\"a\")\n",
        "b = b_idDataFrame.alias(\"b\")\n",
        "\n",
        "final_review_df = a.join(b, col(\"a.business_id\") == col(\"b.business_id\"), 'inner') \\\n",
        "                         .select([col('a.'+xx) for xx in a.columns] + [col('b.businessId')])\n",
        "\n",
        "final_review_df.show(5,truncate=False)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "35Z4g9xeXq_g"
      },
      "source": [
        "rating_df = final_review_df.select('userId', 'businessId', final_review_df.stars.cast('float').alias('rating'))\n",
        "rating_df.show(5)\n",
        "print(' Rating matrx no. of rows :', rating_df.count())\n",
        "rating_df.printSchema()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2bkf7cy2Xq_h"
      },
      "source": [
        "(train,test)=rating_df.randomSplit(weights=[0.8,0.2],seed=123)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8dcKTAAYXq_h"
      },
      "source": [
        "als = ALS(userCol=\"userId\", itemCol=\"businessId\", ratingCol=\"rating\", coldStartStrategy=\"drop\")\n",
        "\n",
        "param_grid = ParamGridBuilder().addGrid(\n",
        "    als.rank,\n",
        "    [10, 15, 20],\n",
        ").addGrid(\n",
        "    als.maxIter,\n",
        "    [10, 15, 20],\n",
        ").build()\n",
        "\n",
        "evaluator = RegressionEvaluator(\n",
        "    metricName=\"rmse\",\n",
        "    labelCol=\"rating\",\n",
        ")\n",
        "\n",
        "cv = CrossValidator(estimator=als, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5, seed=123)\n",
        "cv_als_model = cv.fit(train)\n",
        "\n",
        "# Evaluate the model by computing the RMSE on the test data\n",
        "\n",
        "als_predictions = cv_als_model.bestModel.transform(test)\n",
        "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n",
        "rmse = evaluator.evaluate(als_predictions)\n",
        "print(\"Root-mean-square error = \" + str(rmse))\n",
        "\n",
        "best_model = cv_als_model.bestModel\n",
        "\n",
        "#best_rank is 20\n",
        "best_model.rank\n",
        "\n",
        "#best_maxIter is 20\n",
        "(best_model\n",
        "    ._java_obj     # Get Java object\n",
        "    .parent()      # Get parent (ALS estimator)\n",
        "    .getMaxIter()) # Get maxIter"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Wjyh2DruXq_i"
      },
      "source": [
        "# trying higher rank and maxIter need huge resources. try to tune regParam (default value = 0.1)\n",
        "\n",
        "alsb = ALS(rank=20, maxIter=20, regParam=0.3, userCol=\"userId\", itemCol=\"businessId\", ratingCol=\"rating\", \\\n",
        "               coldStartStrategy=\"drop\", seed=123)\n",
        "alsb_model = alsb.fit(train)\n",
        "\n",
        "alsb_predictions = alsb_model.transform(test)\n",
        "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n",
        "rmse = evaluator.evaluate(alsb_predictions)\n",
        "print(\"Root-mean-square error = \" + str(rmse))\n",
        "\n",
        "# save the ALS model\n",
        "alsb_model.write().overwrite().save(model_path + 'als')\n",
        "\n",
        "# Root-mean-square error is 1.2579688933524986"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "XMEKoLY5Xq_i"
      },
      "source": [
        "# load a new instance of the saved ALS model\n",
        "alsn_model = ALSModel.load(model_path + 'als')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "138RISfGXq_i"
      },
      "source": [
        "userRecoms = alsn_model.recommendForAllUsers(10)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ESXWwWs-Xq_j"
      },
      "source": [
        "# add the column user_id, cache the recommendaton dataframe and show recommedations sample\n",
        "\n",
        "a = userRecoms.alias(\"a\")\n",
        "b = user_newid_df.alias(\"b\")\n",
        "    \n",
        "all_userRecoms = a.join(b, col(\"a.userId\") == col(\"b.userId\"), 'inner') \\\n",
        "             .select([col('a.'+xx) for xx in a.columns] + [col('b.user_id')])\n",
        "\n",
        "all_userRecoms.cache()   \n",
        "all_userRecoms.show(1, truncate=False)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2mlyVZz3Xq_j"
      },
      "source": [
        "u_id = 'ZWD8UH1T7QXQr0Eq-mcWYg'\n",
        "\n",
        "userFlatRec =  sqlContext.createDataFrame(all_userRecoms.filter(col('user_id') == u_id).rdd.flatMap(lambda p: p[1]))\n",
        "userFlatRec.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GDx9NrkGXq_k"
      },
      "source": [
        "a = business_new_df.alias(\"a\")\n",
        "b = userFlatRec.alias(\"b\")\n",
        "\n",
        "user_collab_df = a.join(b, col(\"a.businessId\") == col(\"b.businessId\"), 'inner') \\\n",
        "                         .select([col('a.'+xx) for xx in a.columns] + [col('b.rating')])\n",
        "    \n",
        "user_collab_df.select('business_id', 'business_name', 'rating', 'categories').toPandas()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qvunS7_-Xq_k"
      },
      "source": [
        "def getCollabRecom(u_id):\n",
        "    \n",
        "\n",
        "    userFlatRec =  sqlContext.createDataFrame(all_userRecoms.filter(col('user_id') == u_id).rdd.flatMap(lambda p: p[1]))\n",
        "\n",
        "    a = userFlatRec.alias(\"a\")\n",
        "    b = business_new_df.alias(\"b\")\n",
        "    \n",
        "    return a.join(b, col(\"a.businessId\") == col(\"b.businessId\"), 'inner') \\\n",
        "             .select([col('b.business_id'), col('a.rating'), col('b.business_name'),col('b.categories'),\n",
        "                                                           col('b.stars'),col('b.review_count'),\n",
        "                                                           col('b.latitude'),col('b.longitude')]) \\\n",
        "             .orderBy(\"rating\", ascending = False)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bH5vZ5PAXq_l"
      },
      "source": [
        "u_id = 'ZWD8UH1T7QXQr0Eq-mcWYg'\n",
        "getCollabRecom(u_id).toPandas()"
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}