hotel_etl.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import re\n",
    "import pyspark\n",
    "from pyspark.sql import SQLContext, functions, types\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "from pyspark.sql import Row\n",
    "from geopy.geocoders import Nominatim"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc=pyspark.SparkContext(appName=\"project\")\n",
    "spark = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Reading hotel dataset scraped from TripAdvisor\n",
    "\n",
    "h_df = pd.read_json('tripadvisor_hotel_output/hotel_info.json')\n",
    "h1_df = spark.createDataFrame(h_df).cache()\n",
    "h1_df.createOrReplaceTempView('h1_df')\n",
    "\n",
    "## Removing duplicates from the hotel dataset\n",
    "\n",
    "temp=spark.sql(\"SELECT df.id FROM (SELECT id, COUNT(*) as tot_count FROM h1_df GROUP BY id ORDER BY tot_count DESC) df WHERE df.tot_count>1\")\n",
    "temp.createOrReplaceTempView('temp')\n",
    "del_dup = spark.sql(\"SELECT h1_df.* FROM h1_df LEFT JOIN temp ON h1_df.id == temp.id WHERE temp.id IS NULL\").cache()\n",
    "del_dup.createOrReplaceTempView('del_dup')\n",
    "\n",
    "## Splitting amenities based on ',' and type casting to array of strings\n",
    "\n",
    "del_dup = del_dup.withColumn(\"amenities\", functions.split(del_dup[\"amenities\"], \",\").cast(\"array<string>\"))\n",
    "\n",
    "## Filling missing prices\n",
    "\n",
    "prices = [float(i[0]) for i in del_dup.select(\"price\").dropna().collect() if i[0] != 'NaN']\n",
    "avg_price = sum(prices)/len(prices)\n",
    "avg_price_df = del_dup.withColumn('price',functions.when(functions.isnan(functions.col(\"price\")), functions.lit(avg_price)).otherwise(functions.col(\"price\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Getting city using coordinates\n",
    "\n",
    "geolocator = Nominatim(user_agent=\"new_recomm\", timeout=None)\n",
    "\n",
    "def get_cname(x):\n",
    "    if 'nil' in x[1:-1]:\n",
    "        return \"None\"\n",
    "    else:\n",
    "        location = geolocator.reverse(x[1:-1], timeout=None)\n",
    "        if 'city' in location.raw[\"address\"]:\n",
    "            return location.raw[\"address\"][\"city\"]\n",
    "        elif 'town' in location.raw[\"address\"]:\n",
    "            return location.raw[\"address\"][\"town\"]\n",
    "        else:\n",
    "            return \"None\"\n",
    "    \n",
    "get_city = functions.udf(lambda a:get_cname(a),types.StringType())\n",
    "\n",
    "city_df = avg_price_df.withColumn(\"city\",get_city(functions.col(\"location\"))).cache()\n",
    "\n",
    "## Saving etled dataset\n",
    "\n",
    "city_df.createOrReplaceTempView('del_dup')\n",
    "city_df.coalesce(4).write.json('etl/del_dup',mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Explode amenities to make predictions based on length of amentities provided by user\n",
    "\n",
    "newh_df  = spark.sql(\"SELECT id,explode(amenities) as amenities FROM del_dup\")\n",
    "\n",
    "##  Removing punctuations from amenities column\n",
    "\n",
    "strip_udf = functions.udf(lambda x: re.sub(r'[^\\w\\s]','',x), types.StringType())\n",
    "newh_df = newh_df.withColumn(\"amenities\", strip_udf(functions.col(\"amenities\")))\n",
    "newh_df.createOrReplaceTempView('newh_df')\n",
    "newh_df.coalesce(4).write.json('etl/newh_df',mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_json('tripadvisor_hotel_output/reviews.json')\n",
    "\n",
    "df[\"att_id\"]=df.id.astype('category').cat.codes\n",
    "\n",
    "rev_df = spark.createDataFrame(df).cache()\n",
    "rev_df.createOrReplaceTempView('rev_df')\n",
    "\n",
    "rev_temp=spark.sql(\"SELECT df.id FROM (SELECT id, COUNT(*) as tot_count FROM rev_df GROUP BY id ORDER BY tot_count DESC) df WHERE df.tot_count>1\")\n",
    "rev_temp.createOrReplaceTempView('rev_temp')\n",
    "\n",
    "s_df = spark.sql(\"SELECT rev_df.* FROM rev_df LEFT JOIN rev_temp ON rev_df.id == rev_temp.id WHERE rev_temp.id IS NULL\")\n",
    "s_df.createOrReplaceTempView('s_df')\n",
    "\n",
    "\n",
    "## String Indexing user_name \n",
    "\n",
    "indexer = StringIndexer(inputCol=\"user_name\", outputCol=\"user_id\")\n",
    "indexed = indexer.fit(s_df).transform(s_df)\n",
    "u_id_df = indexed.withColumn(\"user_id\",indexed[\"user_id\"].cast(\"Int\")).cache()\n",
    "u_id_df.createOrReplaceTempView('u_id_df')\n",
    "u_id_df.coalesce(4).write.json('etl/u_id_df',mode='overwrite')"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}