CAPSTONE-PROJECT.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import necessary libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e3e9fc38d48d4d0a8a535cec0e1a169e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>3</td><td>application_1610958643234_0004</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-4-135.us-west-2.compute.internal:20888/proxy/application_1610958643234_0004/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-4-135.us-west-2.compute.internal:8042/node/containerlogs/container_1610958643234_0004_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# we don't necessary need to import SparkSession within the emr jupyter notebook, \n",
    "#only if we want to to turn the code in a python script and use spark-submit\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.config(\"spark.jars.packages\",\"org.apache.hadoop:hadoop-aws:2.7.0\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f99f489639314630b28f599720dfcc3e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import seaborn\n",
    "import numpy\n",
    "import pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "0447f013b87c40fb8e4d4ca0f2db9d5d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.functions import unix_timestamp\n",
    "from pyspark.sql.functions import udf, col, regexp_replace, when, countDistinct, count, asc, desc\n",
    "from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format, to_timestamp\n",
    "from pyspark.sql import types"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Input : calendar ==> Output : time_table & calendar fact table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ebb3092b0a664ccc9b96cbac440d877b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# read the calendar.csv from s3\n",
    "calendar=spark.read.format('csv').options(header='true').load('s3://omar-archive-files/amsterdam-data/calendar.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "28724e3e9ab6483b808cdd91db4e24ad",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# transformation applied to calendar columns \n",
    "\"\"\"\n",
    "Input dataframe : raw dataframe with string-type columns and messed up with symbols like $ sign\n",
    "Output dataframe: refined dataframe with the appropriate column types \"\"\"\n",
    "calendar= calendar.withColumn('new_date', calendar['date'].cast('date'))\\\n",
    "                  .withColumn('new_listing_id', calendar['listing_id'].cast(dataType=types.IntegerType()))\\\n",
    "                  .withColumn('new_price', regexp_replace(col('price'), '[$,]', '').cast('float'))\\\n",
    "                  .withColumn('new_available', (when(col('available') == 'f', 'False').when(col('available') == 't', 'True')).cast(dataType=types.BooleanType()))\n",
    "\n",
    "calendar=calendar.select(col('new_listing_id').alias('listing_id'),\\\n",
    "                         col('new_date').alias('date') ,\\\n",
    "                         col('new_available').alias('available'),\\\n",
    "                         col('new_price').alias('price'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a3384272314c4fff98ca783d5dfbadc7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- listing_id: integer (nullable = true)\n",
      " |-- date: date (nullable = true)\n",
      " |-- available: boolean (nullable = true)\n",
      " |-- price: float (nullable = true)"
     ]
    }
   ],
   "source": [
    "calendar.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e5b1acd338e84b12a775d06bb3dd7cde",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+---------+-----+\n",
      "|listing_id|      date|available|price|\n",
      "+----------+----------+---------+-----+\n",
      "|      2818|2019-12-05|    false| null|\n",
      "|     73208|2019-08-30|    false| null|\n",
      "|     73208|2019-08-29|    false| null|\n",
      "|     73208|2019-08-28|    false| null|\n",
      "|     73208|2019-08-27|    false| null|\n",
      "+----------+----------+---------+-----+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "calendar.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The price column may include null values, however I will not delete it since the null value represents an informative value which is the listing is unvailable and is already taken that night"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3bc1836c15df4d47a897071893d71a8d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7310950"
     ]
    }
   ],
   "source": [
    "calendar.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "83ead263e3114fdeacceea71fb3d56f2",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "20030.0"
     ]
    }
   ],
   "source": [
    "7310950/365"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The calendar dataframe registers 7310950 records referring to 20030 listings * 365 days. \n",
    "So I need a unique key to uniquely identify a row in the calendar dataframe based on the (listing_id,date).\n",
    "The count of this key should be 7310950"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "361c8f245d434c3cbe95ec9ac1fb1346",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "calendar.createOrReplaceTempView('calendar')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f02dd2c5b09d4e3198afc99d54513cee",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "calendar=spark.sql(\"\"\"\n",
    "            select row_number() over \\\n",
    "            (order by (listing_id,date)) as calendar_id,* from calendar\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ccca9efa2b2048ab9782138d4c7d0f5d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "calendar.createOrReplaceTempView('calendar')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ee34e1d9b7814cb585c5b51fa3e940b1",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------+\n",
      "|count(DISTINCT calendar_id)|\n",
      "+---------------------------+\n",
      "|                    7310950|\n",
      "+---------------------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select count(distinct (calendar_id)) from calendar\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cc8ede7d0c524a0599bd8def10365c3f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+----------+---------+-----+\n",
      "|calendar_id|listing_id|      date|available|price|\n",
      "+-----------+----------+----------+---------+-----+\n",
      "|          1|      2818|2018-12-06|     true| 59.0|\n",
      "|          2|      2818|2018-12-07|    false| null|\n",
      "|          3|      2818|2018-12-08|    false| null|\n",
      "|          4|      2818|2018-12-09|    false| null|\n",
      "|          5|      2818|2018-12-10|    false| null|\n",
      "+-----------+----------+----------+---------+-----+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "calendar.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1e7487deb9d54186a0f19e0a6805b307",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "calendar.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"calendar.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e9cce4d53d6a43348d02ea41155c2968",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# extract date_time table from dates in calendar which expand from 06/12/18 to 06/12/19\n",
    "# the date_time table will be complete later as the calendar does not include all needed dates \n",
    "time_table=calendar.select('date').\\\n",
    "dropDuplicates().orderBy('date', ascending=True)\\\n",
    ".withColumn('day_of_month', dayofmonth('date'))\\\n",
    ".withColumn('week'        , weekofyear('date'))\\\n",
    ".withColumn('month'       , month('date'))\\\n",
    ".withColumn('year'        , year('date'))\\\n",
    ".withColumn('day'         , date_format('date', 'EEEE'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e2c64c539225483aae1215d5b85dff64",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "time_table.createOrReplaceTempView('time_table')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "36d3b4251b7e4a5f81351bd8d6e9fd69",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "| min(date)|\n",
      "+----------+\n",
      "|2018-12-06|\n",
      "+----------+"
     ]
    }
   ],
   "source": [
    "spark.sql('select min( date) from time_table').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "01a256e5f71b45df9418adf21c889eac",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "| max(date)|\n",
      "+----------+\n",
      "|2019-12-06|\n",
      "+----------+"
     ]
    }
   ],
   "source": [
    "spark.sql('select max( date) from time_table').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Input : listings ==> Output : location , listings, host\n",
    "#### Merge listings and listings details and apply data clearning and transformation to form listings table, host and location table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "042d1983946d4dd8beed1c9d8c1259c0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings=spark.read.format('csv').options(header='true').load('s3://omar-archive-files/amsterdam-data/listings.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ddf8ffc1316a43c0a07b657004446fe3",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings_details=spark.read.format('csv').option('multiLine', True).options(header='true').option('escape','\"').load('s3://omar-archive-files/amsterdam-data/listings_details.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9c38e6e7fd294a7ba2903976dcb2a766",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# join the two tables as they complement each other and extract maximum info\n",
    "new_listings = listings.alias('a').join(listings_details.alias('b'), on=['id'], how='left_outer')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d402e8d4c378454b9a70d4028316c773",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "the number of the listings is : 20048"
     ]
    }
   ],
   "source": [
    "print (\"the number of the listings is :\", new_listings.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7e4701ead3994ccca4bff940635fbfaf",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# extract only wanted columns and apply cleaning and modification to its values such casting data types.\n",
    "new_listings=new_listings.withColumn('new_price'        , regexp_replace(col('a.price'),'[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_weekly_price' , regexp_replace(col('weekly_price'),'[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_monthly_price', regexp_replace(col('monthly_price'), '[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_security_deposit' , regexp_replace(col('security_deposit'),  '[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_cleaning_fee' , regexp_replace(col('cleaning_fee'),'[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_guests_included'   , regexp_replace(col('guests_included'),'[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_extra_people'     , regexp_replace(col('extra_people'),'[$,]', '').cast('float'))\\\n",
    "            .withColumn('new_require_guest_profile_picture',\\\n",
    "                        (when(col('require_guest_profile_picture') == 'f', 'False').\\\n",
    "                         when(col('require_guest_profile_picture') == 't', 'True')).cast(dataType=types.BooleanType()))\\\n",
    "            .withColumn('new_require_guest_phone_verification',\\\n",
    "                        (when(col('require_guest_phone_verification') == 'f', 'False').\\\n",
    "                         when(col('require_guest_phone_verification') == 't', 'True')).cast(dataType=types.BooleanType()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create location_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "b8174641b46b4eb4a205ab8164a20945",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# for every listing extract its location information and put it into location_table\n",
    "location_table=new_listings.select(col('b.latitude').cast('float'),\\\n",
    "                   col('b.longitude').cast('float'),\\\n",
    "                   col('street'),\\\n",
    "                   col('zipcode'),\\\n",
    "                   col('city'),\\\n",
    "                   col('country') ).where(col('latitude').isNotNull() & col('longitude').isNotNull()).\\\n",
    "                    dropDuplicates(subset=['latitude', 'longitude'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e52f969990a1406788b149257bba1766",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# like the calendar table, location_table needs a unique key to identify certain values for (latitude,longitude)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c0a876fa288249858f6bbe7606ba1ef5",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "location_table.createOrReplaceTempView(\"location\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "43dab3f435de4b35b298e0c8620673c9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "location_table=spark.sql(\"\"\"\n",
    "            select row_number() over \\\n",
    "            (order by (latitude,longitude)) as location_id,* from location\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "fa85f1b2579241ccac0200d1cd77b81c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "location_table.createOrReplaceTempView(\"location_with_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ea65dc6e36ae497cb28e3905935f8000",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|max(location_id)|\n",
      "+----------------+\n",
      "|           20029|\n",
      "+----------------+"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select max (location_id )from location_with_id\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ce418e6e957144ddb829aee1bcdef627",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------+---------+--------------------+-------+-------------------+-----------+\n",
      "|location_id| latitude|longitude|              street|zipcode|               city|    country|\n",
      "+-----------+---------+---------+--------------------+-------+-------------------+-----------+\n",
      "|          1|      2.0|      2.0|\",,,https://a0.mu...|    N/A|     within an hour|    De Pijp|\n",
      "|          2| 52.28838|4.9756355|Amsterdam, Noord-...|1107 VC|          Amsterdam|Netherlands|\n",
      "|          3|52.289276|4.9737587|Amsterdam Zuid-Oo...|1107 VE|Amsterdam Zuid-Oost|Netherlands|\n",
      "|          4|52.290306|4.9784713|Amsterdam-Zuidoos...| 1107WN| Amsterdam-Zuidoost|Netherlands|\n",
      "|          5|52.290688| 4.983784|Amsterdam Zuid-Oo...|   1052|Amsterdam Zuid-Oost|Netherlands|\n",
      "+-----------+---------+---------+--------------------+-------+-------------------+-----------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "location_table.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cefb1420044a44438e2733d2bc17f9c5",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- location_id: integer (nullable = true)\n",
      " |-- latitude: float (nullable = true)\n",
      " |-- longitude: float (nullable = true)\n",
      " |-- street: string (nullable = true)\n",
      " |-- zipcode: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- country: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "location_table.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7d2a0eaba2f74b4f9b451fa4a01491a0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "location_table.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"location_table.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "46c4d4431a4748a382bd01bba83c21ed",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# I am a big fan of pandas, here is a basic demo on how to use it with a spark dataframes\n",
    "# the transformation to pandas dataframe may be useful when a matplotlib/seaborn graph is needed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9cf82ecd7e114b30beddc1a35866d01a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df=location_table.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "efccd65ae5264271a3fa34e50b58fd16",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   location_id   latitude  longitude  ...  zipcode                 city      country\n",
      "0            1   2.000000   2.000000  ...      N/A       within an hour      De Pijp\n",
      "1            2  52.288380   4.975636  ...  1107 VC            Amsterdam  Netherlands\n",
      "2            3  52.289276   4.973759  ...  1107 VE  Amsterdam Zuid-Oost  Netherlands\n",
      "3            4  52.290306   4.978471  ...   1107WN   Amsterdam-Zuidoost  Netherlands\n",
      "4            5  52.290688   4.983784  ...     1052  Amsterdam Zuid-Oost  Netherlands\n",
      "\n",
      "[5 rows x 7 columns]"
     ]
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7e3b5e7cb2fe4f0aa95df0cd6f47748e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Amsterdam, Noord-Holland, Netherlands             15360\n",
      "Amsterdam, North Holland, Netherlands              2899\n",
      "Amsterdam, NH, Netherlands                         1252\n",
      "Amsterdam-Zuidoost, Noord-Holland, Netherlands      156\n",
      "Amsterdam, Netherlands                              120\n",
      "                                                  ...  \n",
      "Amsterdam, Overtoom, Netherlands                      1\n",
      "Amsterdam, boord Holland, Netherlands                 1\n",
      "Amsterdam, CE, Netherlands                            1\n",
      "1015 SN Amsterdam, Noord-Holland, Netherlands         1\n",
      "Amsterdam, LN, Netherlands                            1\n",
      "Name: street, Length: 104, dtype: int64"
     ]
    }
   ],
   "source": [
    "df['street'].value_counts()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "2b8a50e3f614492a831f422084783b81",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#here I finish editing my listings_table ( ordering & renaming columns, type casting)\n",
    "listings_table=new_listings.select(col('id').cast('integer'),\\\n",
    "                   col('a.host_id').cast('integer'),\\\n",
    "                   col('a.latitude').cast('float'),\\\n",
    "                   col('a.longitude').cast('float'),\\\n",
    "                    \n",
    "                   col('new_price').alias('price'),\\\n",
    "                   col('new_weekly_price').alias('weekly_price'),\\\n",
    "                   col('new_monthly_price').alias('monthly_price'),\\\n",
    "                   col('new_security_deposit').alias('security_deposit'),\\\n",
    "                   col('new_cleaning_fee').alias('cleaning_fee'),\\\n",
    "                   col('new_guests_included').alias('guests_included'),\\\n",
    "                   col('new_extra_people').alias('extra_people'),\\\n",
    "\n",
    "                   col('listing_url'),\\\n",
    "\n",
    "                   col('property_type'),\\\n",
    "                   col('accommodates').cast('integer'),\\\n",
    "                   col('a.room_type'),\\\n",
    "                   col('bathrooms').cast('float'),\\\n",
    "                   col('bedrooms').cast('float'),\\\n",
    "                   col('beds').cast('float'),\\\n",
    "                   col('bed_type'),\\\n",
    "                   col('amenities'),\\\n",
    "\n",
    "                   col('review_scores_rating').cast('float'),\\\n",
    "                   col('review_scores_accuracy').cast('float'),\\\n",
    "                   col('review_scores_cleanliness').cast('float'),\\\n",
    "                   col('review_scores_location').cast('float'),\\\n",
    "                   col('review_scores_value').cast('float'),\\\n",
    "                   col('cancellation_policy'),\\\n",
    "                   col('new_require_guest_profile_picture').alias('require_guest_profile_picture'),\\\n",
    "                   col('new_require_guest_phone_verification').alias('require_guest_phone_verification'))           "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### So far my listings_table contains pretty much columns, and the idea of setting the location info apart is good. So I'll create a location_id in the listings_table to references the the location_id within the location_table. This link is based on the values of (latitude, longitude).. so no need to explicitly put them within listings_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "bd6359e8c5f84693b488b823cf5e27e3",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings_table.createOrReplaceTempView('listings')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d74ec8a9671446bbb6c7679c218bbb3e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings_table=spark.sql(\"\"\" select * , loc.location_id from listings as lis \\\n",
    "                                         INNER JOIN location_with_id as loc\\\n",
    "                                         ON (lis.latitude = loc.latitude AND lis.longitude = loc.longitude)\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dd3b81c8dd924bb18aa09ff5ca97c482",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings_table=listings_table.select(\"id\", \"host_id\", \"location_id\", \"price\", \\\n",
    "                      \"weekly_price\", \"monthly_price\", \"security_deposit\", \\\n",
    "                     \"cleaning_fee\", \"guests_included\", \"extra_people\",\\\n",
    "                     \"listing_url\", \"property_type\", \"accommodates\", \\\n",
    "                     \"room_type\", \"bathrooms\", \"bedrooms\", \\\n",
    "                      \"beds\", \"bed_type\", \"amenities\", \\\n",
    "                      \"review_scores_rating\", \"review_scores_accuracy\", \"review_scores_cleanliness\",\\\n",
    "                      \"review_scores_location\",\"review_scores_value\", \"cancellation_policy\",\\\n",
    "                      \"require_guest_profile_picture\",\"require_guest_phone_verification\").where(col(\"id\").isNotNull())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f371f0a391474f0ba77eceee1ab6aba7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: integer (nullable = true)\n",
      " |-- host_id: integer (nullable = true)\n",
      " |-- location_id: integer (nullable = true)\n",
      " |-- price: float (nullable = true)\n",
      " |-- weekly_price: float (nullable = true)\n",
      " |-- monthly_price: float (nullable = true)\n",
      " |-- security_deposit: float (nullable = true)\n",
      " |-- cleaning_fee: float (nullable = true)\n",
      " |-- guests_included: float (nullable = true)\n",
      " |-- extra_people: float (nullable = true)\n",
      " |-- listing_url: string (nullable = true)\n",
      " |-- property_type: string (nullable = true)\n",
      " |-- accommodates: integer (nullable = true)\n",
      " |-- room_type: string (nullable = true)\n",
      " |-- bathrooms: float (nullable = true)\n",
      " |-- bedrooms: float (nullable = true)\n",
      " |-- beds: float (nullable = true)\n",
      " |-- bed_type: string (nullable = true)\n",
      " |-- amenities: string (nullable = true)\n",
      " |-- review_scores_rating: float (nullable = true)\n",
      " |-- review_scores_accuracy: float (nullable = true)\n",
      " |-- review_scores_cleanliness: float (nullable = true)\n",
      " |-- review_scores_location: float (nullable = true)\n",
      " |-- review_scores_value: float (nullable = true)\n",
      " |-- cancellation_policy: string (nullable = true)\n",
      " |-- require_guest_profile_picture: boolean (nullable = true)\n",
      " |-- require_guest_phone_verification: boolean (nullable = true)"
     ]
    }
   ],
   "source": [
    "listings_table.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a18ad5d0802a4b49bd88dbc66d4544aa",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "listings_table.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"listings.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## host table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### now return back to the new_listings dataframe to extract the host dimension table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f7c24b3697cd41f9a3964efb476217f8",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "new_listings=new_listings.withColumn('new_host_response_rate',regexp_replace(col('host_response_rate'),'[$,%]', '' ).cast('float'))\\\n",
    "            .withColumn('new_host_is_superhost',\\\n",
    "                        (when(col('host_is_superhost') == 'f', 'False').\\\n",
    "                         when(col('host_is_superhost') == 't', 'True')).cast(dataType=types.BooleanType()))\\\n",
    "            .withColumn('new_host_identity_verified',\\\n",
    "                        (when(col('host_identity_verified') == 'f', 'False').\\\n",
    "                         when(col('host_identity_verified') == 't', 'True')).cast(dataType=types.BooleanType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1a2b7380c027494ba5e20d0f60eb4417",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "host_table=new_listings.select(col('b.host_id').cast('integer'),\\\n",
    "                              col('b.host_name'),\\\n",
    "                              col('host_about'),\\\n",
    "                              col('host_response_time'),\\\n",
    "                              col('new_host_response_rate').alias('host_response_rate'),\\\n",
    "                              col('new_host_is_superhost').alias('host_is_superhost'),\\\n",
    "                              col('host_location'),\\\n",
    "                              col('new_host_identity_verified').alias('host_identity_verified'))\\\n",
    "                              .where(col(\"host_id\").isNotNull()).dropDuplicates(subset=['host_id'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "01b617286f114adeb79665865e3d613c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "17263"
     ]
    }
   ],
   "source": [
    "host_table.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "2a7e87c6d619467389a0d8f91b9c406d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- host_id: integer (nullable = true)\n",
      " |-- host_name: string (nullable = true)\n",
      " |-- host_about: string (nullable = true)\n",
      " |-- host_response_time: string (nullable = true)\n",
      " |-- host_response_rate: float (nullable = true)\n",
      " |-- host_is_superhost: boolean (nullable = true)\n",
      " |-- host_location: string (nullable = true)\n",
      " |-- host_identity_verified: boolean (nullable = true)"
     ]
    }
   ],
   "source": [
    "host_table.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "5b193027adb7464697c00905deb95200",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "host_table.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"hosts.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## reviews"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a772aa3a39f74dd085b1ed6dc16c08ae",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# read in a json format file of reviews_details \n",
    "reviews_details=spark.read.format('json').option(\"multiLine\",\"true\").load('s3://omar-archive-files/amsterdam-data/reviews_details.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7f05913775ce4e67bdafcc165561bca7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+----+----------+-----------+-------------+\n",
      "|            comments|      date|  id|listing_id|reviewer_id|reviewer_name|\n",
      "+--------------------+----------+----+----------+-----------+-------------+\n",
      "|Daniel is really ...|2009-03-30|1191|      2818|      10952|          Lam|\n",
      "|Daniel is the mos...|2009-04-24|1771|      2818|      12798|        Alice|\n",
      "|We had such a gre...|2009-05-03|1989|      2818|      11869|      Natalja|\n",
      "|Very professional...|2009-05-18|2797|      2818|      14064|      Enrique|\n",
      "|Daniel is highly ...|2009-05-25|3151|      2818|      17977|      Sherwin|\n",
      "+--------------------+----------+----+----------+-----------+-------------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "reviews_details.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Do remember that our time_table previously defined starts up from 06/12/2018, however reviews have been made since 2009. Here below I join dates outside 2018-2019 to have a date_time table which covers all the AirBnB events"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c0d4b523865d4b9a91d43bf88e474385",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "reviews_details=reviews_details.withColumn('new_date', reviews_details['date'].cast('date'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "537693d2566d4909bfad4d20db2f569c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "time_table2=reviews_details.selectExpr('new_date as date').dropDuplicates().where(col('date').isNotNull()).orderBy('date', ascending=True)\\\n",
    ".withColumn('day_of_month', dayofmonth('date'))\\\n",
    ".withColumn('week'        , weekofyear('date'))\\\n",
    ".withColumn('month'       , month('date'))\\\n",
    ".withColumn('year'        , year('date'))\\\n",
    ".withColumn('day'         , date_format('date', 'EEEE'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9ca2c12eba2f4828ad095098f1afbd1b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "date_time= time_table2.union(time_table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "56230a2d4e354b98ad54cb4e9058d482",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- date: date (nullable = true)\n",
      " |-- day_of_month: integer (nullable = true)\n",
      " |-- week: integer (nullable = true)\n",
      " |-- month: integer (nullable = true)\n",
      " |-- year: integer (nullable = true)\n",
      " |-- day: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "date_time.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "522af55bf05a4394880badc265ba40f9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+----+-----+----+------+\n",
      "|      date|day_of_month|week|month|year|   day|\n",
      "+----------+------------+----+-----+----+------+\n",
      "|2009-03-30|          30|  14|    3|2009|Monday|\n",
      "|2009-04-24|          24|  17|    4|2009|Friday|\n",
      "|2009-05-03|           3|  18|    5|2009|Sunday|\n",
      "|2009-05-18|          18|  21|    5|2009|Monday|\n",
      "|2009-05-25|          25|  22|    5|2009|Monday|\n",
      "+----------+------------+----+-----+----+------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "date_time.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "4375c4c6b1f84f45b222233c9e3d55bb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " date_time starts from 2009 now and actually has 3240 records"
     ]
    }
   ],
   "source": [
    "print ( \" date_time starts from 2009 now and actually has {} records\".format(date_time.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "43aa48f818f34658a0644c148d9b28ed",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "date_time.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"date_time.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f0d57d4d6e0e46a880e434f4a80f4504",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "## reviewers_table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### in addition to calendar fact table, my hybrid star schema or snowflake schema would involve another fact table which is reviews ( completely indep on the first fact table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dfcabcbcdb9d469181abeb350f1e8cf2",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "fact_reviews=reviews_details.select(col('id').cast('integer'),\\\n",
    "                                    col('listing_id').cast('integer'),\\\n",
    "                                    col('new_date').alias('date'),\\\n",
    "                                    col('reviewer_id').cast('integer'),\\\n",
    "                                    col('comments')).where(col('id').isNotNull() & col('date').isNotNull() & col('reviewer_id').isNotNull())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e531c748138f48019069fea73c28c997",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: integer (nullable = true)\n",
      " |-- listing_id: integer (nullable = true)\n",
      " |-- date: date (nullable = true)\n",
      " |-- reviewer_id: integer (nullable = true)\n",
      " |-- comments: string (nullable = true)"
     ]
    }
   ],
   "source": [
    "fact_reviews.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "415c2a70fddd4c1e97e2a004ac761801",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "fact_reviews.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"fact_reviews.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Finaly I add reviewer dimension table and to detach the name of the reviewer from the fact table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f4ea2e7cab9748dbbf5df77a25d0636b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "reviewer_table=reviews_details.select(col('reviewer_id').cast('integer'), \\\n",
    "                                     col('reviewer_name')).where(col(\"reviewer_id\").isNotNull()).dropDuplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "60a478c1503b4b38a57b353b3684f18e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "reviewer_table.write.parquet(\"s3://omar-archive-files/amsterdam-data/\"+\"reviewer_table.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a305cf2b307e4f748548ce73ce6ca9c9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------+\n",
      "|reviewer_id|reviewer_name|\n",
      "+-----------+-------------+\n",
      "|    1475799|       Dhaval|\n",
      "|   17837628|       Sophie|\n",
      "|   60449351|      Etienne|\n",
      "|  101794231|        Katie|\n",
      "|   26289366|       Martin|\n",
      "+-----------+-------------+\n",
      "only showing top 5 rows"
     ]
    }
   ],
   "source": [
    "reviewer_table.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "039b744309c948159c4d1057665d77e8",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There is 411813 active reviewers in the data set"
     ]
    }
   ],
   "source": [
    "print(\"There is {} active reviewers in the data set\".format(reviewer_table.dropDuplicates().count()))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}