Sparkify Data Lake.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 0. Prepare working environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 287,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import necessary libraries\n",
    "from datetime import datetime\n",
    "import os\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf, col\n",
    "from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format\n",
    "import pandas as pd\n",
    "from pyspark.sql.types import *\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create local spark session\n",
    "spark = SparkSession.builder\\\n",
    "                    .master(\"local\")\\\n",
    "                    .appName(\"SparkifyMusic\")\\\n",
    "                    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.1.10:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.3</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>SparkifyMusic</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f2cf83e06a0>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Checking spark session details\n",
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('spark.master', 'local'),\n",
       " ('spark.rdd.compress', 'True'),\n",
       " ('spark.serializer.objectStreamReset', '100'),\n",
       " ('spark.app.name', 'SparkifyMusic'),\n",
       " ('spark.app.id', 'local-1561176073104'),\n",
       " ('spark.executor.id', 'driver'),\n",
       " ('spark.submit.deployMode', 'client'),\n",
       " ('spark.driver.host', '192.168.1.10'),\n",
       " ('spark.ui.showConsoleProgress', 'true'),\n",
       " ('spark.driver.port', '39495')]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Checking spark session config details\n",
    "spark.sparkContext.getConf().getAll()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 309,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define paths\n",
    "output_path = \"./output_data/\"\n",
    "song_data = \"./data/song-data/song_data/*/*/*/*.json\"\n",
    "log_data = \"./data/log-data/*.json\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Process Song Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.1 Create and store songs table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 316,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist_id: string (nullable = true)\n",
      " |-- artist_latitude: double (nullable = true)\n",
      " |-- artist_location: string (nullable = true)\n",
      " |-- artist_longitude: double (nullable = true)\n",
      " |-- artist_name: string (nullable = true)\n",
      " |-- duration: double (nullable = true)\n",
      " |-- num_songs: long (nullable = true)\n",
      " |-- song_id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- year: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Read song data\n",
    "songs_df = spark.read.json(song_data)\n",
    "songs_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 317,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Redefine schema to enforce the constraints (if needed, part of cleansing step)\n",
    "# This is based on the needs of specific business requirements\n",
    "songs_schema = StructType([\n",
    "    StructField(\"artist_id\", StringType(), False),\n",
    "    StructField(\"artist_latitude\", DoubleType(), True),\n",
    "    StructField(\"artist_location\", StringType(), True),\n",
    "    StructField(\"artist_longitude\", DoubleType(), True),\n",
    "    StructField(\"artist_name\", StringType(), True),\n",
    "    StructField(\"duration\", DoubleType(), True),\n",
    "    StructField(\"num_songs\", LongType(), True),\n",
    "    StructField(\"song_id\", StringType(), False),\n",
    "    StructField(\"title\", StringType(), True),\n",
    "    StructField(\"year\", IntegerType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 318,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reload the songs_df after apply the pre-defined schema struct type\n",
    "# Any violated rows will be droped during loading\n",
    "songs_df = spark.read.json(song_data,\n",
    "                    schema=songs_schema,\n",
    "                    mode=\"DROPMALFORMED\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 319,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist_id: string (nullable = true)\n",
      " |-- artist_latitude: double (nullable = true)\n",
      " |-- artist_location: string (nullable = true)\n",
      " |-- artist_longitude: double (nullable = true)\n",
      " |-- artist_name: string (nullable = true)\n",
      " |-- duration: double (nullable = true)\n",
      " |-- num_songs: long (nullable = true)\n",
      " |-- song_id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- year: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the new schema\n",
    "songs_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 320,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temp view for querying\n",
    "songs_df.createOrReplaceTempView(\"songs\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 321,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>song_id</th>\n",
       "      <th>title</th>\n",
       "      <th>artist_id</th>\n",
       "      <th>year</th>\n",
       "      <th>duration</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>SOBAYLL12A8C138AF9</td>\n",
       "      <td>Sono andati? Fingevo di dormire</td>\n",
       "      <td>ARDR4AC1187FB371A1</td>\n",
       "      <td>0</td>\n",
       "      <td>511.16363</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>SOOLYAZ12A6701F4A6</td>\n",
       "      <td>Laws Patrolling (Album Version)</td>\n",
       "      <td>AREBBGV1187FB523D2</td>\n",
       "      <td>0</td>\n",
       "      <td>173.66159</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>SOBBUGU12A8C13E95D</td>\n",
       "      <td>Setting Fire to Sleeping Giants</td>\n",
       "      <td>ARMAC4T1187FB3FA4C</td>\n",
       "      <td>2004</td>\n",
       "      <td>207.77751</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>SOAOIBZ12AB01815BE</td>\n",
       "      <td>I Hold Your Hand In Mine [Live At Royal Albert Hall]</td>\n",
       "      <td>ARPBNLO1187FB3D52F</td>\n",
       "      <td>2000</td>\n",
       "      <td>43.36281</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>SONWXQJ12A8C134D94</td>\n",
       "      <td>The Ballad Of Sleeping Beauty</td>\n",
       "      <td>ARNF6401187FB57032</td>\n",
       "      <td>1994</td>\n",
       "      <td>305.16200</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "              song_id                                                 title  \\\n",
       "0  SOBAYLL12A8C138AF9                       Sono andati? Fingevo di dormire   \n",
       "1  SOOLYAZ12A6701F4A6                       Laws Patrolling (Album Version)   \n",
       "2  SOBBUGU12A8C13E95D                       Setting Fire to Sleeping Giants   \n",
       "3  SOAOIBZ12AB01815BE  I Hold Your Hand In Mine [Live At Royal Albert Hall]   \n",
       "4  SONWXQJ12A8C134D94                         The Ballad Of Sleeping Beauty   \n",
       "\n",
       "            artist_id  year   duration  \n",
       "0  ARDR4AC1187FB371A1     0  511.16363  \n",
       "1  AREBBGV1187FB523D2     0  173.66159  \n",
       "2  ARMAC4T1187FB3FA4C  2004  207.77751  \n",
       "3  ARPBNLO1187FB3D52F  2000   43.36281  \n",
       "4  ARNF6401187FB57032  1994  305.16200  "
      ]
     },
     "execution_count": 321,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get songs table\n",
    "songs_table = spark.sql(\"\"\"\n",
    "            SELECT song_id, \n",
    "            title,\n",
    "            artist_id,\n",
    "            year,\n",
    "            duration\n",
    "            FROM songs\n",
    "            \"\"\")\n",
    "songs_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 322,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Store songs table to disk by partition by year and artist_id\n",
    "songs_table.write.mode(\"overwrite\").partitionBy(\"year\", \"artist_id\").parquet( output_path + \"songs\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.2 Create and store artists table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 323,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>artist_id</th>\n",
       "      <th>name</th>\n",
       "      <th>location</th>\n",
       "      <th>latitude</th>\n",
       "      <th>longitude</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARPBNLO1187FB3D52F</td>\n",
       "      <td>Tiny Tim</td>\n",
       "      <td>New York, NY</td>\n",
       "      <td>40.71455</td>\n",
       "      <td>-74.00712</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>ARBEBBY1187B9B43DB</td>\n",
       "      <td>Tom Petty</td>\n",
       "      <td>Gainesville, FL</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>AR0IAWL1187B9A96D0</td>\n",
       "      <td>Danilo Perez</td>\n",
       "      <td>Panama</td>\n",
       "      <td>8.41770</td>\n",
       "      <td>-80.11278</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>ARMBR4Y1187B9990EB</td>\n",
       "      <td>David Martin</td>\n",
       "      <td>California - SF</td>\n",
       "      <td>37.77916</td>\n",
       "      <td>-122.42005</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>ARD0S291187B9B7BF5</td>\n",
       "      <td>Rated R</td>\n",
       "      <td>Ohio</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            artist_id          name         location  latitude  longitude\n",
       "0  ARPBNLO1187FB3D52F      Tiny Tim     New York, NY  40.71455  -74.00712\n",
       "1  ARBEBBY1187B9B43DB     Tom Petty  Gainesville, FL       NaN        NaN\n",
       "2  AR0IAWL1187B9A96D0  Danilo Perez           Panama   8.41770  -80.11278\n",
       "3  ARMBR4Y1187B9990EB  David Martin  California - SF  37.77916 -122.42005\n",
       "4  ARD0S291187B9B7BF5       Rated R             Ohio       NaN        NaN"
      ]
     },
     "execution_count": 323,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get artist table from songs temp view\n",
    "artists_table = spark.sql(\"\"\"\n",
    "                SELECT DISTINCT artist_id,\n",
    "                artist_name name,\n",
    "                artist_location location,\n",
    "                artist_latitude latitude,\n",
    "                artist_longitude longitude\n",
    "                FROM songs\n",
    "                \"\"\")\n",
    "artists_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 324,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Store artists and save to disks\n",
    "artists_table.write.mode(\"overwrite\").parquet(output_path + \"artists\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Process log data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.1 Read log data and prepare data to build \"users\", \"time\" & \"songplays\" table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 325,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: double (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Read logs and parse to spark dataframe\n",
    "logs_df = spark.read.json(log_data)\n",
    "logs_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 327,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>artist</th>\n",
       "      <th>auth</th>\n",
       "      <th>firstName</th>\n",
       "      <th>gender</th>\n",
       "      <th>itemInSession</th>\n",
       "      <th>lastName</th>\n",
       "      <th>length</th>\n",
       "      <th>level</th>\n",
       "      <th>location</th>\n",
       "      <th>method</th>\n",
       "      <th>page</th>\n",
       "      <th>registration</th>\n",
       "      <th>sessionId</th>\n",
       "      <th>song</th>\n",
       "      <th>status</th>\n",
       "      <th>ts</th>\n",
       "      <th>userAgent</th>\n",
       "      <th>userId</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Harmonia</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Smith</td>\n",
       "      <td>655.77751</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>Sehr kosmisch</td>\n",
       "      <td>200</td>\n",
       "      <td>1542241826796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>The Prodigy</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>1</td>\n",
       "      <td>Smith</td>\n",
       "      <td>260.07465</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>The Big Gundown</td>\n",
       "      <td>200</td>\n",
       "      <td>1542242481796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Train</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>2</td>\n",
       "      <td>Smith</td>\n",
       "      <td>205.45261</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>Marry Me</td>\n",
       "      <td>200</td>\n",
       "      <td>1542242741796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Sony Wonder</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Samuel</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Gonzalez</td>\n",
       "      <td>218.06975</td>\n",
       "      <td>free</td>\n",
       "      <td>Houston-The Woodlands-Sugar Land, TX</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.540493e+12</td>\n",
       "      <td>597</td>\n",
       "      <td>Blackbird</td>\n",
       "      <td>200</td>\n",
       "      <td>1542253449796</td>\n",
       "      <td>\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/7.0.5 Safari/537.77.4\"</td>\n",
       "      <td>61</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Van Halen</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Tegan</td>\n",
       "      <td>F</td>\n",
       "      <td>2</td>\n",
       "      <td>Levine</td>\n",
       "      <td>289.38404</td>\n",
       "      <td>paid</td>\n",
       "      <td>Portland-South Portland, ME</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.540794e+12</td>\n",
       "      <td>602</td>\n",
       "      <td>Best Of Both Worlds (Remastered Album Version)</td>\n",
       "      <td>200</td>\n",
       "      <td>1542260935796</td>\n",
       "      <td>\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"</td>\n",
       "      <td>80</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        artist       auth firstName gender  itemInSession  lastName  \\\n",
       "0     Harmonia  Logged In      Ryan      M              0     Smith   \n",
       "1  The Prodigy  Logged In      Ryan      M              1     Smith   \n",
       "2        Train  Logged In      Ryan      M              2     Smith   \n",
       "3  Sony Wonder  Logged In    Samuel      M              0  Gonzalez   \n",
       "4    Van Halen  Logged In     Tegan      F              2    Levine   \n",
       "\n",
       "      length level                              location method      page  \\\n",
       "0  655.77751  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "1  260.07465  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "2  205.45261  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "3  218.06975  free  Houston-The Woodlands-Sugar Land, TX    PUT  NextSong   \n",
       "4  289.38404  paid           Portland-South Portland, ME    PUT  NextSong   \n",
       "\n",
       "   registration  sessionId                                            song  \\\n",
       "0  1.541017e+12        583                                   Sehr kosmisch   \n",
       "1  1.541017e+12        583                                 The Big Gundown   \n",
       "2  1.541017e+12        583                                        Marry Me   \n",
       "3  1.540493e+12        597                                       Blackbird   \n",
       "4  1.540794e+12        602  Best Of Both Worlds (Remastered Album Version)   \n",
       "\n",
       "   status             ts  \\\n",
       "0     200  1542241826796   \n",
       "1     200  1542242481796   \n",
       "2     200  1542242741796   \n",
       "3     200  1542253449796   \n",
       "4     200  1542260935796   \n",
       "\n",
       "                                                                                                                                   userAgent  \\\n",
       "0  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "1  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "2  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "3                    \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/7.0.5 Safari/537.77.4\"   \n",
       "4                 \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"   \n",
       "\n",
       "  userId  \n",
       "0     26  \n",
       "1     26  \n",
       "2     26  \n",
       "3     61  \n",
       "4     80  "
      ]
     },
     "execution_count": 327,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Filter page of NextSong only\n",
    "logs_df = logs_df.filter(logs_df[\"page\"] == \"NextSong\")\n",
    "logs_df.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 328,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temp view used for SQL query\n",
    "logs_df.createOrReplaceTempView(\"logs\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.2 Create and store users table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 330,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_id</th>\n",
       "      <th>first_name</th>\n",
       "      <th>last_name</th>\n",
       "      <th>gender</th>\n",
       "      <th>level</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>98</td>\n",
       "      <td>Jordyn</td>\n",
       "      <td>Powell</td>\n",
       "      <td>F</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>34</td>\n",
       "      <td>Evelin</td>\n",
       "      <td>Ayala</td>\n",
       "      <td>F</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>85</td>\n",
       "      <td>Kinsley</td>\n",
       "      <td>Young</td>\n",
       "      <td>F</td>\n",
       "      <td>paid</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>38</td>\n",
       "      <td>Gianna</td>\n",
       "      <td>Jones</td>\n",
       "      <td>F</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>85</td>\n",
       "      <td>Kinsley</td>\n",
       "      <td>Young</td>\n",
       "      <td>F</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  user_id first_name last_name gender level\n",
       "0      98     Jordyn    Powell      F  free\n",
       "1      34     Evelin     Ayala      F  free\n",
       "2      85    Kinsley     Young      F  paid\n",
       "3      38     Gianna     Jones      F  free\n",
       "4      85    Kinsley     Young      F  free"
      ]
     },
     "execution_count": 330,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get user table from logs\n",
    "users_table = spark.sql(\"\"\"\n",
    "                    SELECT DISTINCT userId user_id,\n",
    "                    firstName first_name,\n",
    "                    lastName last_name,\n",
    "                    gender,\n",
    "                    level\n",
    "                    FROM logs\n",
    "                    WHERE TRIM(userId) <> ''\n",
    "                \"\"\")\n",
    "users_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 331,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Store user table\n",
    "users_table.write.mode(\"overwrite\").parquet(output_path + \"users\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.3 Create and store time table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 337,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>artist</th>\n",
       "      <th>auth</th>\n",
       "      <th>firstName</th>\n",
       "      <th>gender</th>\n",
       "      <th>itemInSession</th>\n",
       "      <th>lastName</th>\n",
       "      <th>length</th>\n",
       "      <th>level</th>\n",
       "      <th>location</th>\n",
       "      <th>method</th>\n",
       "      <th>page</th>\n",
       "      <th>registration</th>\n",
       "      <th>sessionId</th>\n",
       "      <th>song</th>\n",
       "      <th>status</th>\n",
       "      <th>ts</th>\n",
       "      <th>userAgent</th>\n",
       "      <th>userId</th>\n",
       "      <th>timestamp</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Harmonia</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Smith</td>\n",
       "      <td>655.77751</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>Sehr kosmisch</td>\n",
       "      <td>200</td>\n",
       "      <td>1542241826796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 07:30:26.796</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>The Prodigy</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>1</td>\n",
       "      <td>Smith</td>\n",
       "      <td>260.07465</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>The Big Gundown</td>\n",
       "      <td>200</td>\n",
       "      <td>1542242481796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 07:41:21.796</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Train</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>M</td>\n",
       "      <td>2</td>\n",
       "      <td>Smith</td>\n",
       "      <td>205.45261</td>\n",
       "      <td>free</td>\n",
       "      <td>San Jose-Sunnyvale-Santa Clara, CA</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.541017e+12</td>\n",
       "      <td>583</td>\n",
       "      <td>Marry Me</td>\n",
       "      <td>200</td>\n",
       "      <td>1542242741796</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 07:45:41.796</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Sony Wonder</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Samuel</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Gonzalez</td>\n",
       "      <td>218.06975</td>\n",
       "      <td>free</td>\n",
       "      <td>Houston-The Woodlands-Sugar Land, TX</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.540493e+12</td>\n",
       "      <td>597</td>\n",
       "      <td>Blackbird</td>\n",
       "      <td>200</td>\n",
       "      <td>1542253449796</td>\n",
       "      <td>\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/7.0.5 Safari/537.77.4\"</td>\n",
       "      <td>61</td>\n",
       "      <td>2018-11-15 10:44:09.796</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Van Halen</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Tegan</td>\n",
       "      <td>F</td>\n",
       "      <td>2</td>\n",
       "      <td>Levine</td>\n",
       "      <td>289.38404</td>\n",
       "      <td>paid</td>\n",
       "      <td>Portland-South Portland, ME</td>\n",
       "      <td>PUT</td>\n",
       "      <td>NextSong</td>\n",
       "      <td>1.540794e+12</td>\n",
       "      <td>602</td>\n",
       "      <td>Best Of Both Worlds (Remastered Album Version)</td>\n",
       "      <td>200</td>\n",
       "      <td>1542260935796</td>\n",
       "      <td>\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"</td>\n",
       "      <td>80</td>\n",
       "      <td>2018-11-15 12:48:55.796</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        artist       auth firstName gender  itemInSession  lastName  \\\n",
       "0     Harmonia  Logged In      Ryan      M              0     Smith   \n",
       "1  The Prodigy  Logged In      Ryan      M              1     Smith   \n",
       "2        Train  Logged In      Ryan      M              2     Smith   \n",
       "3  Sony Wonder  Logged In    Samuel      M              0  Gonzalez   \n",
       "4    Van Halen  Logged In     Tegan      F              2    Levine   \n",
       "\n",
       "      length level                              location method      page  \\\n",
       "0  655.77751  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "1  260.07465  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "2  205.45261  free    San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong   \n",
       "3  218.06975  free  Houston-The Woodlands-Sugar Land, TX    PUT  NextSong   \n",
       "4  289.38404  paid           Portland-South Portland, ME    PUT  NextSong   \n",
       "\n",
       "   registration  sessionId                                            song  \\\n",
       "0  1.541017e+12        583                                   Sehr kosmisch   \n",
       "1  1.541017e+12        583                                 The Big Gundown   \n",
       "2  1.541017e+12        583                                        Marry Me   \n",
       "3  1.540493e+12        597                                       Blackbird   \n",
       "4  1.540794e+12        602  Best Of Both Worlds (Remastered Album Version)   \n",
       "\n",
       "   status             ts  \\\n",
       "0     200  1542241826796   \n",
       "1     200  1542242481796   \n",
       "2     200  1542242741796   \n",
       "3     200  1542253449796   \n",
       "4     200  1542260935796   \n",
       "\n",
       "                                                                                                                                   userAgent  \\\n",
       "0  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "1  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "2  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "3                    \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/7.0.5 Safari/537.77.4\"   \n",
       "4                 \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"   \n",
       "\n",
       "  userId               timestamp  \n",
       "0     26 2018-11-15 07:30:26.796  \n",
       "1     26 2018-11-15 07:41:21.796  \n",
       "2     26 2018-11-15 07:45:41.796  \n",
       "3     61 2018-11-15 10:44:09.796  \n",
       "4     80 2018-11-15 12:48:55.796  "
      ]
     },
     "execution_count": 337,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Add in timestamp column for current logs table\n",
    "get_timestamp = udf(lambda x: datetime.fromtimestamp(x/1000), TimestampType())\n",
    "logs_df = logs_df.withColumn(\"timestamp\", get_timestamp(logs_df.ts))\n",
    "logs_df.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 333,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: double (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the schema\n",
    "logs_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 301,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temp view for timestamp for query\n",
    "logs_df.select(\"timestamp\").createOrReplaceTempView(\"time\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 334,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>start_time</th>\n",
       "      <th>hour</th>\n",
       "      <th>day</th>\n",
       "      <th>week</th>\n",
       "      <th>month</th>\n",
       "      <th>year</th>\n",
       "      <th>weekday</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2018-11-15 13:07:37.796</td>\n",
       "      <td>13</td>\n",
       "      <td>15</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2018-11-15 19:30:36.796</td>\n",
       "      <td>19</td>\n",
       "      <td>15</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2018-11-15 21:49:51.796</td>\n",
       "      <td>21</td>\n",
       "      <td>15</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2018-11-15 23:06:06.796</td>\n",
       "      <td>23</td>\n",
       "      <td>15</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2018-11-16 04:14:54.796</td>\n",
       "      <td>4</td>\n",
       "      <td>16</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               start_time  hour  day  week  month  year  weekday\n",
       "0 2018-11-15 13:07:37.796    13   15    46     11  2018        5\n",
       "1 2018-11-15 19:30:36.796    19   15    46     11  2018        5\n",
       "2 2018-11-15 21:49:51.796    21   15    46     11  2018        5\n",
       "3 2018-11-15 23:06:06.796    23   15    46     11  2018        5\n",
       "4 2018-11-16 04:14:54.796     4   16    46     11  2018        6"
      ]
     },
     "execution_count": 334,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get time table from user's event logs\n",
    "time_table = spark.sql(\"\"\"\n",
    "SELECT DISTINCT timestamp start_time,\n",
    "HOUR(timestamp) hour,\n",
    "DAYOFMONTH(timestamp) day,\n",
    "WEEKOFYEAR(timestamp) week,\n",
    "MONTH(timestamp) month,\n",
    "YEAR(timestamp) year,\n",
    "DAYOFWEEK(timestamp) weekday\n",
    "FROM time\n",
    "\"\"\")\n",
    "time_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 304,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write time table to disk with partitioning by year and month\n",
    "time_table.write.mode(\"overwrite\").partitionBy(\"year\", \"month\").parquet(output_path + \"time\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.4 Create and store songplays_table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 338,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temp view used for SQL query\n",
    "logs_df.createOrReplaceTempView(\"logs\")\n",
    "\n",
    "# Load songs table\n",
    "songs_df = spark.read.parquet(output_path + \"songs\")\n",
    "songs_df.createOrReplaceTempView(\"songs\")\n",
    "\n",
    "# Load artists table\n",
    "artists_df = spark.read.parquet(output_path + \"artists\")\n",
    "artists_df.createOrReplaceTempView(\"artists\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 350,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>songplay_id</th>\n",
       "      <th>start_time</th>\n",
       "      <th>user_id</th>\n",
       "      <th>song_id</th>\n",
       "      <th>artist_id</th>\n",
       "      <th>session_id</th>\n",
       "      <th>location</th>\n",
       "      <th>user_agent</th>\n",
       "      <th>month</th>\n",
       "      <th>year</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>2018-11-22 04:56:47.796</td>\n",
       "      <td>15</td>\n",
       "      <td>SOZCTXZ12AB0182364</td>\n",
       "      <td>AR5KOSW1187FB35FF4</td>\n",
       "      <td>818</td>\n",
       "      <td>Chicago-Naperville-Elgin, IL-IN-WI</td>\n",
       "      <td>\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   songplay_id              start_time user_id             song_id  \\\n",
       "0            0 2018-11-22 04:56:47.796      15  SOZCTXZ12AB0182364   \n",
       "\n",
       "            artist_id  session_id                            location  \\\n",
       "0  AR5KOSW1187FB35FF4         818  Chicago-Naperville-Elgin, IL-IN-WI   \n",
       "\n",
       "                                                                                                                                  user_agent  \\\n",
       "0  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/36.0.1985.125 Chrome/36.0.1985.125 Safari/537.36\"   \n",
       "\n",
       "   month  year  \n",
       "0     11  2018  "
      ]
     },
     "execution_count": 350,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create songplays table\n",
    "songplays_table = spark.sql(\"\"\"\n",
    "SELECT monotonically_increasing_id() songplay_id,\n",
    "l.timestamp start_time,\n",
    "l.userId user_id,\n",
    "s.song_id,\n",
    "a.artist_id,\n",
    "l.sessionId session_id,\n",
    "l.location,\n",
    "l.userAgent user_agent,\n",
    "MONTH(l.timestamp) month,\n",
    "YEAR(l.timestamp) year\n",
    "FROM logs l\n",
    "INNER JOIN songs s ON s.title = l.song AND s.duration = l.length\n",
    "INNER JOIN artists a ON a.artist_id = s.artist_id AND a.name = l.artist\n",
    "\"\"\")\n",
    "songplays_table.limit(10).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 351,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write songplays table to disk\n",
    "songplays_table.write.mode(\"overwrite\").partitionBy(\"year\", \"month\").parquet(output_path + \"songplays\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}