yelp_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import seaborn as sns\n",
    "from pyspark import SparkConf,SparkContext\n",
    "import matplotlib.pyplot as plt\n",
    "from time import time \n",
    "from tqdm import tqdm\n",
    "import sklearn\n",
    "from pyspark.ml.feature import HashingTF, IDF, Tokenizer,StringIndexer\n",
    "from pyspark.sql.functions import col,udf\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql.functions import col, countDistinct\n",
    "from pyspark.ml import Pipeline,PipelineModel\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.mllib.evaluation import BinaryClassificationMetrics,MulticlassMetrics\n",
    "from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"yelp-data-exploration\") \\\n",
    "    .config(\"spark.driver.maxResultSize\", \"100g\") \\\n",
    "    .config(\"spark.driver.memory\", \"50g\") \\\n",
    "    .config(\"spark.executor.memory\", \"20g\") \\\n",
    "    .config(\"spark.master\", \"local[2]\") \\\n",
    "    .config(\"spark.local.dir\",\"/home/bh_parijat/spark_temp/\")\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#sc = SparkSession.builder.appName('yelp-data-exploration').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users = sc.read.json(\"/home/bh_parijat/data/yelp/user.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "yelp_reviews = sc.read.json(\"/home/bh_parijat/data/yelp/review.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_business = sc.read.json(\"/home/bh_parijat/data/yelp/business.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip = sc.read.json(\"/home/bh_parijat/data/yelp/tip.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6685900"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yelp_reviews.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "reviews_columns = [\"business_id\", \"cool\",\"date\",\"funny\",\"review_id\",\"stars\",\"text\",\"useful\",\"user_id\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "yelp_reviews_df = yelp_reviews.toDF(*reviews_columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(business_id='ujmEBvifdJM6h6RLv4wQIg', cool=0, date='2013-05-07 04:34:36', funny=1, review_id='Q1sbwvVQXV2734tPgoKj4Q', stars=1.0, text='Total bill for this horrible service? Over $8Gs. These crooks actually had the nerve to charge us $69 for 3 pills. I checked online the pills can be had for 19 cents EACH! Avoid Hospital ERs at all costs.', useful=6, user_id='hG7b0MtEbXx5QzbzE6C_VA')]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yelp_reviews.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(business_id='ujmEBvifdJM6h6RLv4wQIg', cool=0, date='2013-05-07 04:34:36', funny=1, review_id='Q1sbwvVQXV2734tPgoKj4Q', stars=1.0, text='Total bill for this horrible service? Over $8Gs. These crooks actually had the nerve to charge us $69 for 3 pills. I checked online the pills can be had for 19 cents EACH! Avoid Hospital ERs at all costs.', useful=6, user_id='hG7b0MtEbXx5QzbzE6C_VA')"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yelp_reviews_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "#users_columns = yelp_users.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "#users_columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users_df = yelp_users.toDF(*users_columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_users_df.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "#business_columns = yelp_business.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_business.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "#business_columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_checkin = sc.read.json(\"/home/bh_parijat/data/yelp/checkin.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#checkin_columns = yelp_checkin.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#checkin_columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_checkin.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_checkin.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_business.agg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_business.agg(countDistinct(col(\"business_id\")).alias(\"different_business\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_checkin.agg(countDistinct(col(\"business_id\")).alias(\"different_business\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_business.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip.agg(countDistinct(col(\"user_id\")).alias(\"different_users\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_tip.agg(countDistinct(col(\"business_id\")).alias(\"business_id\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# yelp_tip.groupBy(\"user_id\")\\\n",
    "#         .agg({'compliment_count':'count'})\\\n",
    "#         .withColumnRenamed(\"count(compliment_count)\",\"count\")\\\n",
    "#         .sort(col(\"count\").desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# yelp_tip.groupBy(\"business_id\")\\\n",
    "#         .agg({'compliment_count':'count'})\\\n",
    "#         .withColumnRenamed(\"count(compliment_count)\",\"count\")\\\n",
    "#         .sort(col(\"count\").desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_reviews"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_reviews.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#r = yelp_reviews.select(\"text\").limit(2).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#r[1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#yelp_reviews.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "# yelp_reviews.groupBy(\"stars\")\\\n",
    "#             .agg({\"text\":\"count\"})\\\n",
    "#             .withColumnRenamed(\"count(text)\",\"count\")\\\n",
    "#             .sort(col(\"count\").desc())\\\n",
    "#             .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## TF-IDF Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "yelp_reviews_df.createOrReplaceTempView(\"reviews\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(stars=1.0, text='Total bill for this horrible service? Over $8Gs. These crooks actually had the nerve to charge us $69 for 3 pills. I checked online the pills can be had for 19 cents EACH! Avoid Hospital ERs at all costs.')]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.sql(\"select stars, text from reviews\").take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def map_rating_sentiment(rating):\n",
    "    return 1 if rating>=3 else 0\n",
    "\n",
    "\n",
    "map_rating = udf(map_rating_sentiment,IntegerType())\n",
    "yelp_reviews_df_final = yelp_reviews_df.withColumn(\"sentiment\",map_rating(\"stars\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+-----+\n",
      "|sentiment|                text|stars|\n",
      "+---------+--------------------+-----+\n",
      "|        0|Total bill for th...|  1.0|\n",
      "|        1|I *adore* Travis ...|  5.0|\n",
      "+---------+--------------------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "yelp_reviews_df_final.select([\"sentiment\",\"text\",\"stars\"]).show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "yelp_reviews_df_final.createOrReplaceTempView(\"reviews\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "X = sc.sql(\"select text,sentiment from reviews\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "X = X.withColumnRenamed(\"text\",\"review\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenizer = Tokenizer(inputCol=\"review\",outputCol=\"tokenized_text\")\n",
    "hashingTF = HashingTF(inputCol=\"tokenized_text\",outputCol=\"mapped_text\")\n",
    "idf = IDF(inputCol=\"mapped_text\",outputCol=\"features\")\n",
    "stringIndexer = StringIndexer(inputCol=\"sentiment\",outputCol=\"label\")\n",
    "lr = LogisticRegression(maxIter=1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline(stages=[tokenizer,hashingTF,idf,stringIndexer,lr])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "#pipelineFit = pipeline.fit(X)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "#X = pipelineFit.transform(X)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------+\n",
      "|              review|sentiment|\n",
      "+--------------------+---------+\n",
      "|Total bill for th...|        0|\n",
      "+--------------------+---------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "X.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "train,test = X.randomSplit([0.85,0.15])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "parameters = ParamGridBuilder()\\\n",
    "            .addGrid(lr.regParam ,[0.001,0.01,0.1,1,10])\\\n",
    "            .build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "tvs = TrainValidationSplit(estimator=pipeline,\n",
    "                           evaluator=BinaryClassificationEvaluator(),\n",
    "                            estimatorParamMaps= parameters,\n",
    "                           trainRatio=0.8\n",
    "                          )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = tvs.fit(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction = model.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction_df = prediction.select(\"features\",\"label\",\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+----------+\n",
      "|            features|label|prediction|\n",
      "+--------------------+-----+----------+\n",
      "|(262144,[9639,244...|  0.0|       0.0|\n",
      "|(262144,[991,1109...|  0.0|       0.0|\n",
      "|(262144,[3067,344...|  0.0|       0.0|\n",
      "|(262144,[1469,271...|  0.0|       0.0|\n",
      "|(262144,[733,1469...|  0.0|       0.0|\n",
      "|(262144,[118,836,...|  1.0|       1.0|\n",
      "|(262144,[3367,963...|  0.0|       0.0|\n",
      "|(262144,[14,9616,...|  0.0|       0.0|\n",
      "|(262144,[14,1998,...|  0.0|       0.0|\n",
      "|(262144,[2666,847...|  0.0|       0.0|\n",
      "+--------------------+-----+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prediction_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "def compare_label(x,y):\n",
    "    return 1 if x==y else 0\n",
    "\n",
    "\n",
    "compare = udf(compare_label,IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "prediction_df = prediction_df.withColumn(\"correct\",compare(\"label\",\"prediction\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "metrics = BinaryClassificationMetrics(prediction_df.select(\"label\",\"prediction\").rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.6693000053079718"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "metrics.areaUnderPR"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Confidence of predicting our classes correct depicted by Area under ROC is 90.6%"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf_metrics  = MulticlassMetrics(prediction_df.select(\"label\",\"prediction\").rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.911682838116951"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_metrics.precision()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.911682838116951"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_metrics.recall()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.911682838116951"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_metrics.accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseMatrix(2, 2, [753200.0, 18046.0, 70509.0, 160938.0], 0)"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_metrics.confusionMatrix()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.911682838116951"
      ]
     },
     "execution_count": 73,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf_metrics.fMeasure()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.bestModel.save(\"model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [],
   "source": [
    "model1 = PipelineModel.load(\"model/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}