YelpAnalysisSpark2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Yelp Data Analysis in Spark 2\n",
    "\n",
    "This is a continuation of the previous notebook YelpAnalysisSpark1.ipnyb which can be found in this repository.\n",
    "This notebook deals whith coming two files business.json and reviews.json to find average reviews from January to May in all the years for food categories near Toronto.\n",
    "\n",
    "First we need to filter out the food businesses near Toronto with business Id and name which will be our left table\n",
    "\n",
    "Second we need to find the reviews given from January to May.\n",
    "\n",
    "Then join the two tables based on Business id \n",
    "\n",
    "Perform analysis on the joined table!\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark import SparkContext as sc\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from pyspark.sql import SQLContext\n",
    "import json\n",
    "import pyspark\n",
    "data_business = \"../Data/business.json\"\n",
    "data_review = \"../Data/review.json\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc = sc(appName=\"Yelp\")\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Load business data\n",
    "testJsonData = sqlContext.read.json(data_business)\n",
    "\n",
    "#Filter the attributes you want\n",
    "LatLong = testJsonData.select(testJsonData.business_id, testJsonData.name, \n",
    "                              pyspark.sql.functions.explode(testJsonData.categories).alias(\"category\"), \n",
    "                              testJsonData.latitude, testJsonData.longitude, testJsonData.stars, \n",
    "                              testJsonData.review_count)\n",
    "\n",
    "#Register Temp Table\n",
    "LatLong.registerTempTable(\"LatLong\")\n",
    "\n",
    "#Then run a SQL query to filter out the ones only the ones within 15km distance from Toronto Center\n",
    "BusinessNearTorontofilt = sqlContext.sql(\"SELECT * FROM  LatLong WHERE acos(sin(0.763782941288) * ",
    "sin(LatLong.latitude * 3.14 /180) + cos(0.763782941288) * ",
    "cos(LatLong.latitude * 3.14 /180) * cos((LatLong.longitude * ",
    "3.14 /180) - (-1.38598479111))) * 6371 <= 15\")\n",
    "BusinessNearTorontofilt.registerTempTable(\"BusinessNearToronto\")\n",
    "\n",
    "#Then filter the business with food category\n",
    "AggQuery = sqlContext.sql(\"SELECT business_id, name FROM BusinessNearToronto WHERE category = 'Food'\")\n",
    "AggQuery.registerTempTable(\"RightTable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|         business_id|                name|\n",
      "+--------------------+--------------------+\n",
      "|v2WhjAB3PIBA8J8Vx...|    The Tea Emporium|\n",
      "|L1XHTn7S-6har9UGA...|Papa Chang's Tea ...|\n",
      "|kEq7eudoX5qdcaSLA...|Paris Bakery & Pa...|\n",
      "|UZShf6G75npKCCjiH...|Fortinos Supermarket|\n",
      "|8gzRJEmYJGTCdahRK...|       Tea & Kitchen|\n",
      "|SRg6VOJLidr8Pw-Im...|Freshii Rutherfor...|\n",
      "|zcWit_aSGR5wiunYB...|          Second Cup|\n",
      "|o-0Nen1h78jiWZF_A...|  Toronto Distillery|\n",
      "|N-FKBizx_wu3L8mvD...|Just Desserts Res...|\n",
      "|nL_IofJ0sQVigmj3H...| Welcome Food Market|\n",
      "|YHXczxm4W3BkGT-z7...|Chatime - Yonge &...|\n",
      "|HnbWx7Q8P4-MXbxMp...|      Jimmy's Coffee|\n",
      "|rBBYRHZk2Bs5IW-GL...|     Columbus Bakery|\n",
      "|opnw3jgcHUsfHCv6Y...|  Aroma Espresso Bar|\n",
      "|ExoqgA3aAz6bYidI0...|California Sandwi...|\n",
      "|k1Nm4agS92jbY8ZKl...|     Pat East Market|\n",
      "|PgtOActam5PdtWn45...|   Bruno's Valu-Mart|\n",
      "|Cr9yQp1qrYOJTx1J-...|Seara Bakery & Pa...|\n",
      "|FiRAbcgD7ja2suCv8...|     Crawford Market|\n",
      "|dlLG3yMa9UtU5K0Vj...|  Purdys Chocolatier|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#We want only those business id which are in 15km radius of toronto with food category\n",
    "AggQuery.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Load Review Data\n",
    "reviewData = sqlContext.read.json(data_review)\n",
    "#Select Business, star  and month attributes\n",
    "reviewDataattr = reviewData.select(reviewData.business_id, reviewData.stars, \n",
    "                                   pyspark.sql.functions.month(reviewData.date).alias(\"month\"))\n",
    "reviewDataattr.registerTempTable(\"reviewData\")\n",
    "#Filter the months before May\n",
    "DateFilter = sqlContext.sql(\"SELECT business_id, stars FROM  reviewData WHERE month < 06\")\n",
    "DateFilter.registerTempTable(\"LeftTable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "JoinedTable = sqlContext.sql(\"SELECT LeftTable.business_id, LeftTable.stars, RightTable.name FROM",
    "LeftTable LEFT JOIN RightTable ON LeftTable.business_id = RightTable.business_id\")\n",
    "JoinedTable.registerTempTable(\"JoinedTable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+----+\n",
      "|         business_id|stars|name|\n",
      "+--------------------+-----+----+\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    2|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    3|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    3|null|\n",
      "|--9e1ONYQuAa-CB_R...|    1|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    2|null|\n",
      "|--9e1ONYQuAa-CB_R...|    4|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    2|null|\n",
      "|--9e1ONYQuAa-CB_R...|    4|null|\n",
      "|--9e1ONYQuAa-CB_R...|    1|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "|--9e1ONYQuAa-CB_R...|    4|null|\n",
      "|--9e1ONYQuAa-CB_R...|    5|null|\n",
      "+--------------------+-----+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "JoinedTable.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "AggResult_Bottom_10 = sqlContext.sql(\"SELECT name, AVG(stars) as Stars FROM JoinedTable",
    "                                     GROUP BY name ORDER BY Stars LIMIT 10\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|                name|Stars|\n",
      "+--------------------+-----+\n",
      "|    Sweet Fix Bakery|  1.0|\n",
      "|     85 C Sugar Time|  1.0|\n",
      "|        Castle Fruit|  1.0|\n",
      "|  Alfredos Fine Food|  1.0|\n",
      "|    Pfannkuchen Köln|  1.0|\n",
      "|           Taco Bell|  1.0|\n",
      "|     House of Cheung|  1.0|\n",
      "|Northwest Kitchen...|  1.0|\n",
      "|      G's Fine Foods|  1.0|\n",
      "|               Doria|  1.0|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "AggResult_Bottom_10.show()\n",
    "AggResult_Bottom_10.coalesce(1).write.format('com.databricks.spark.csv').option(\"header\", \"true\").save('Question5_bot')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "AggResult_Top_10 = sqlContext.sql(\"SELECT name, AVG(stars) as Stars FROM JoinedTable GROUP BY name",
    "                                  ORDER BY Stars DESC LIMIT 10\")"
   ]
  }
 ],
 "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.6.0"
  },
  "name": "Question 5",
  "notebookId": 1342456922744315
 },
 "nbformat": 4,
 "nbformat_minor": 1
}