Test.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "0bbd80b6",
   "metadata": {},
   "source": [
    "https://knowledge.udacity.com/questions/481917 <br>\n",
    "https://knowledge.udacity.com/questions/340141"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9da48aa7",
   "metadata": {},
   "source": [
    "https://knowledge.udacity.com/questions/460708 <br>\n",
    "https://discussions.apple.com/thread/3242258 "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "47af17ee",
   "metadata": {},
   "source": [
    "https://knowledge.udacity.com/questions/825541 <br>\n",
    "https://knowledge.udacity.com/questions/713520"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77ea3a31",
   "metadata": {},
   "source": [
    "# Sparkify DataLake Project\n",
    "BY: MAHMOUD NAGY - OCT 2022\n",
    "\n",
    "- In this Notebook, we are going to work locally on our project with a smaller dataset (found in the workspace), \n",
    "- and then we can move on to the bigger dataset on AWS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "0e8900b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# import configparser\n",
    "from time import time\n",
    "from datetime import datetime\n",
    "import os\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import udf, col, desc, substring\n",
    "from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format\n",
    "from pyspark.sql.types import TimestampType"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba7291fe",
   "metadata": {},
   "source": [
    "## Make sure that your AWS credentials are loaded as env vars\n",
    "We are working locally in this notebook, so we will not use these credentials until we move to the cloud!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ce4a347d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# config = configparser.ConfigParser()\n",
    "# config.read('dl.cfg')\n",
    "\n",
    "# os.environ['AWS_ACCESS_KEY_ID']=config['AWS_ACCESS_KEY_ID']\n",
    "# os.environ['AWS_SECRET_ACCESS_KEY']=config['AWS_SECRET_ACCESS_KEY']"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bc6c2702",
   "metadata": {},
   "source": [
    "## Create Spark Session with the hadoop-aws package\n",
    "hadoop-aws is a library which is used for us to connect with Amazon S3 <br>\n",
    "We are working locally in this notebook, so we don't need to connect with S3 until we move to the cloud!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "94a5b85a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# def create_spark_session():\n",
    "#     spark = SparkSession \\\n",
    "#         .builder \\\n",
    "#         .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0\") \\\n",
    "#         .getOrCreate()\n",
    "#     return spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "89485aac",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/10/12 04:04:48 WARN Utils: Your hostname, Mahmouds-MacBook-Pro.local resolves to a loopback address: 127.0.0.1; using 192.168.1.2 instead (on interface en0)\n",
      "22/10/12 04:04:48 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "22/10/12 04:04:49 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "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.2:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.2.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>TestBeforeCLoud</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fce57345550>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName(\"TestBeforeCLoud\") \\\n",
    "        .getOrCreate()\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11fb7253",
   "metadata": {},
   "source": [
    "- SparkContext configuration cannot be modified on runtime. You have to stop existing context first.\n",
    "- SQLContext configuration can be modified on runtime."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15e9d625",
   "metadata": {},
   "source": [
    "https://stackoverflow.com/questions/32233575/read-all-files-in-a-nested-folder-in-spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "083adfbc",
   "metadata": {},
   "source": [
    "# Song Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "7cac1849",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "71\n",
      "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"
     ]
    },
    {
     "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>artist_latitude</th>\n",
       "      <th>artist_location</th>\n",
       "      <th>artist_longitude</th>\n",
       "      <th>artist_name</th>\n",
       "      <th>duration</th>\n",
       "      <th>num_songs</th>\n",
       "      <th>song_id</th>\n",
       "      <th>title</th>\n",
       "      <th>year</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARDR4AC1187FB371A1</td>\n",
       "      <td>NaN</td>\n",
       "      <td></td>\n",
       "      <td>NaN</td>\n",
       "      <td>Montserrat Caballé;Placido Domingo;Vicente Sar...</td>\n",
       "      <td>511.16363</td>\n",
       "      <td>1</td>\n",
       "      <td>SOBAYLL12A8C138AF9</td>\n",
       "      <td>Sono andati? Fingevo di dormire</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>AREBBGV1187FB523D2</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Houston, TX</td>\n",
       "      <td>NaN</td>\n",
       "      <td>Mike Jones (Featuring CJ_ Mello &amp; Lil' Bran)</td>\n",
       "      <td>173.66159</td>\n",
       "      <td>1</td>\n",
       "      <td>SOOLYAZ12A6701F4A6</td>\n",
       "      <td>Laws Patrolling (Album Version)</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>ARMAC4T1187FB3FA4C</td>\n",
       "      <td>40.82624</td>\n",
       "      <td>Morris Plains, NJ</td>\n",
       "      <td>-74.47995</td>\n",
       "      <td>The Dillinger Escape Plan</td>\n",
       "      <td>207.77751</td>\n",
       "      <td>1</td>\n",
       "      <td>SOBBUGU12A8C13E95D</td>\n",
       "      <td>Setting Fire to Sleeping Giants</td>\n",
       "      <td>2004</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>ARPBNLO1187FB3D52F</td>\n",
       "      <td>40.71455</td>\n",
       "      <td>New York, NY</td>\n",
       "      <td>-74.00712</td>\n",
       "      <td>Tiny Tim</td>\n",
       "      <td>43.36281</td>\n",
       "      <td>1</td>\n",
       "      <td>SOAOIBZ12AB01815BE</td>\n",
       "      <td>I Hold Your Hand In Mine [Live At Royal Albert...</td>\n",
       "      <td>2000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>ARDNS031187B9924F0</td>\n",
       "      <td>32.67828</td>\n",
       "      <td>Georgia</td>\n",
       "      <td>-83.22295</td>\n",
       "      <td>Tim Wilson</td>\n",
       "      <td>186.48771</td>\n",
       "      <td>1</td>\n",
       "      <td>SONYPOM12A8C13B2D7</td>\n",
       "      <td>I Think My Wife Is Running Around On Me (Taco ...</td>\n",
       "      <td>2005</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            artist_id  artist_latitude    artist_location  artist_longitude  \\\n",
       "0  ARDR4AC1187FB371A1              NaN                                  NaN   \n",
       "1  AREBBGV1187FB523D2              NaN        Houston, TX               NaN   \n",
       "2  ARMAC4T1187FB3FA4C         40.82624  Morris Plains, NJ         -74.47995   \n",
       "3  ARPBNLO1187FB3D52F         40.71455       New York, NY         -74.00712   \n",
       "4  ARDNS031187B9924F0         32.67828            Georgia         -83.22295   \n",
       "\n",
       "                                         artist_name   duration  num_songs  \\\n",
       "0  Montserrat Caballé;Placido Domingo;Vicente Sar...  511.16363          1   \n",
       "1       Mike Jones (Featuring CJ_ Mello & Lil' Bran)  173.66159          1   \n",
       "2                          The Dillinger Escape Plan  207.77751          1   \n",
       "3                                           Tiny Tim   43.36281          1   \n",
       "4                                         Tim Wilson  186.48771          1   \n",
       "\n",
       "              song_id                                              title  year  \n",
       "0  SOBAYLL12A8C138AF9                    Sono andati? Fingevo di dormire     0  \n",
       "1  SOOLYAZ12A6701F4A6                    Laws Patrolling (Album Version)     0  \n",
       "2  SOBBUGU12A8C13E95D                    Setting Fire to Sleeping Giants  2004  \n",
       "3  SOAOIBZ12AB01815BE  I Hold Your Hand In Mine [Live At Royal Albert...  2000  \n",
       "4  SONYPOM12A8C13B2D7  I Think My Wife Is Running Around On Me (Taco ...  2005  "
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# get filepath to song data file\n",
    "song_data = \"data/song_data\"\n",
    "    \n",
    "# read song data file\n",
    "df_songs = spark.read\\\n",
    "          .option(\"recursiveFileLookup\",\"true\")\\\n",
    "          .json(song_data)\n",
    "\n",
    "print(df_songs.count())\n",
    "df_songs.printSchema()\n",
    "df_songs.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3ed9f431",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e43598b7",
   "metadata": {},
   "source": [
    "## Songs Table"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d694e767",
   "metadata": {},
   "source": [
    "## Extract Columns Using Spark Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2f97e08e",
   "metadata": {},
   "source": [
    "song_id, title, artist_id, year, duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "a698a901",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.0606839656829834\n",
      "71\n",
      "root\n",
      " |-- song_id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- artist_id: string (nullable = true)\n",
      " |-- year: long (nullable = true)\n",
      " |-- duration: double (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>SOTTDKS12AB018D69B</td>\n",
       "      <td>It Wont Be Christmas</td>\n",
       "      <td>ARMBR4Y1187B9990EB</td>\n",
       "      <td>0</td>\n",
       "      <td>241.47546</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</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>2</th>\n",
       "      <td>SONYPOM12A8C13B2D7</td>\n",
       "      <td>I Think My Wife Is Running Around On Me (Taco ...</td>\n",
       "      <td>ARDNS031187B9924F0</td>\n",
       "      <td>2005</td>\n",
       "      <td>186.48771</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>SOYMRWW12A6D4FAB14</td>\n",
       "      <td>The Moon And I (Ordinary Day Album Version)</td>\n",
       "      <td>ARKFYS91187B98E58F</td>\n",
       "      <td>0</td>\n",
       "      <td>267.70240</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>SOAOIBZ12AB01815BE</td>\n",
       "      <td>I Hold Your Hand In Mine [Live At Royal Albert...</td>\n",
       "      <td>ARPBNLO1187FB3D52F</td>\n",
       "      <td>2000</td>\n",
       "      <td>43.36281</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "              song_id                                              title  \\\n",
       "0  SOTTDKS12AB018D69B                               It Wont Be Christmas   \n",
       "1  SOBBUGU12A8C13E95D                    Setting Fire to Sleeping Giants   \n",
       "2  SONYPOM12A8C13B2D7  I Think My Wife Is Running Around On Me (Taco ...   \n",
       "3  SOYMRWW12A6D4FAB14        The Moon And I (Ordinary Day Album Version)   \n",
       "4  SOAOIBZ12AB01815BE  I Hold Your Hand In Mine [Live At Royal Albert...   \n",
       "\n",
       "            artist_id  year   duration  \n",
       "0  ARMBR4Y1187B9990EB     0  241.47546  \n",
       "1  ARMAC4T1187FB3FA4C  2004  207.77751  \n",
       "2  ARDNS031187B9924F0  2005  186.48771  \n",
       "3  ARKFYS91187B98E58F     0  267.70240  \n",
       "4  ARPBNLO1187FB3D52F  2000   43.36281  "
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "t0 = time()\n",
    "# extract columns to create songs table\n",
    "songs_table = df_songs.select([\"song_id\", \"title\", \"artist_id\", \"year\", \"duration\"]).dropDuplicates()\n",
    "\n",
    "print(time()-t0)\n",
    "print(songs_table.count())\n",
    "songs_table.printSchema()\n",
    "songs_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6bc57a62",
   "metadata": {},
   "source": [
    "## Extract Columns Using Spark SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "c9275750",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.07294607162475586\n",
      "71\n",
      "root\n",
      " |-- song_id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- artist_id: string (nullable = true)\n",
      " |-- year: long (nullable = true)\n",
      " |-- duration: double (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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...</td>\n",
       "      <td>ARPBNLO1187FB3D52F</td>\n",
       "      <td>2000</td>\n",
       "      <td>43.36281</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>SONYPOM12A8C13B2D7</td>\n",
       "      <td>I Think My Wife Is Running Around On Me (Taco ...</td>\n",
       "      <td>ARDNS031187B9924F0</td>\n",
       "      <td>2005</td>\n",
       "      <td>186.48771</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...   \n",
       "4  SONYPOM12A8C13B2D7  I Think My Wife Is Running Around On Me (Taco ...   \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  ARDNS031187B9924F0  2005  186.48771  "
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_songs.createOrReplaceTempView(\"songs\")\n",
    "t0 = time()\n",
    "# extract columns to create songs table\n",
    "songs_table = spark.sql(\"\"\"\n",
    "    SELECT song_id, title, artist_id, year, duration\n",
    "    FROM songs\n",
    "\"\"\")\n",
    "\n",
    "print(time()-t0)\n",
    "print(songs_table.count())\n",
    "songs_table.printSchema()\n",
    "songs_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "abce59eb",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d8d53147",
   "metadata": {},
   "source": [
    "#### Apache Parquet \n",
    "- is an open source, column-oriented data file format designed for efficient data storage and retrieval."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "4dcfeb61",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# write songs table to parquet files partitioned by year and artist\n",
    "songs_table.write\\\n",
    ".partitionBy(\"year\", \"artist_id\")\\\n",
    ".mode('overwrite')\\\n",
    ".parquet('./songs')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "4e2eb11d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # write songs table to parquet files partitioned by year and artist\n",
    "# song_data.write\\\n",
    "# .partitionBy(\"year\", \"artist_id\")\\\n",
    "# .mode('overwrite')\\\n",
    "# .parquet(os.path.join('s3a://udacity-dend-output/', 'songs'))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0c0480f8",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5c0f918",
   "metadata": {},
   "source": [
    "https://towardsdatascience.com/data-skew-in-pyspark-783d529a9dd7"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a78b703",
   "metadata": {},
   "source": [
    "#### 1. Repartition by Column(s)\n",
    "- The first solution is to logically re-partition your data based on the transformations in your script. \n",
    "- In short, if you’re grouping or joining, partitioning by the groupBy/join columns can improve shuffle efficiency.\n",
    "- `df = df.repartition(<n_partitions>, '<col_1>', '<col_2>',...)`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f4eedd45",
   "metadata": {},
   "source": [
    "- `repartition()` is used to partition data in memory \n",
    "- and `partitionBy` is used to partition data on disk. \n",
    "- They're often used in conjunction.\n",
    "<br>\n",
    "- `repartition()` is used for shuffling data, that's why it occurs in memory.\n",
    "- Both `repartition()` and `partitionBy` can be used to \"partition data based on dataframe column\", \n",
    "- but `repartition()` partitions the data in memory and `partitionBy` partitions the data on disk."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c3e5dc7",
   "metadata": {},
   "source": [
    "## 2 Different Ways to solve the skewed data problem\n",
    "- Assign a new, temporary partition key before processing any huge shuffles\n",
    "- The 2nd method is using `repartition()`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24311a49",
   "metadata": {},
   "source": [
    "The files are partitioned by the first three letters of each song's track ID. <br>\n",
    "**For example, here are filepaths to two files in this dataset.** <br>\n",
    "`song_data/A/B/C/TRABCEI128F424C983.json` <br>\n",
    "`song_data/A/A/B/TRAABJL12903CDCF1A.json`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "dd2d7da2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "e711bb82",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 1 master, 4 cores --> since my local machine is core i5 --> meaning that it has 5 cores for processing data\n",
    "sc.defaultParallelism "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "2c591f89",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|year|count|\n",
      "+----+-----+\n",
      "|   0|   43|\n",
      "|1961|    1|\n",
      "|1964|    1|\n",
      "|1969|    1|\n",
      "|1972|    1|\n",
      "|1982|    1|\n",
      "|1984|    1|\n",
      "|1985|    1|\n",
      "|1986|    1|\n",
      "|1987|    1|\n",
      "|1992|    1|\n",
      "|1993|    1|\n",
      "|1994|    2|\n",
      "|1997|    2|\n",
      "|1999|    1|\n",
      "|2000|    2|\n",
      "|2003|    2|\n",
      "|2004|    4|\n",
      "|2005|    2|\n",
      "|2007|    1|\n",
      "+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_songs.groupBy('year').count().sort('year').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "281ce806",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+-----+\n",
      "|         artist_id|count|\n",
      "+------------------+-----+\n",
      "|ARNTLGG11E2835DDB9|    2|\n",
      "|ARD7TVE1187B99BFB1|    2|\n",
      "|AR9AWNF1187B9AB0B4|    1|\n",
      "|AR0RCMP1187FB3F427|    1|\n",
      "|AR10USD1187B99F3F1|    1|\n",
      "|ARB29H41187B98F0EF|    1|\n",
      "|ARMAC4T1187FB3FA4C|    1|\n",
      "|AREBBGV1187FB523D2|    1|\n",
      "|ARKFYS91187B98E58F|    1|\n",
      "|ARLTWXK1187FB5A3F8|    1|\n",
      "|ARPFHN61187FB575F6|    1|\n",
      "|ARDR4AC1187FB371A1|    1|\n",
      "|AR47JEX1187B995D81|    1|\n",
      "|ARPBNLO1187FB3D52F|    1|\n",
      "|ARQGYP71187FB44566|    1|\n",
      "|ARMBR4Y1187B9990EB|    1|\n",
      "|AROUOZZ1187B9ABE51|    1|\n",
      "|ARNF6401187FB57032|    1|\n",
      "|ARI2JSK1187FB496EF|    1|\n",
      "|ARDNS031187B9924F0|    1|\n",
      "+------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_songs.groupBy('artist_id').count().sort(desc('count')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "1c76cd65",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+-----+\n",
      "|year|         artist_id|count|\n",
      "+----+------------------+-----+\n",
      "|   0|ARD7TVE1187B99BFB1|    2|\n",
      "|   0|ARNTLGG11E2835DDB9|    2|\n",
      "|   0|ARPFHN61187FB575F6|    1|\n",
      "|   0|ARI2JSK1187FB496EF|    1|\n",
      "|2000|ARPBNLO1187FB3D52F|    1|\n",
      "|1992|AR0RCMP1187FB3F427|    1|\n",
      "|   0|ARMBR4Y1187B9990EB|    1|\n",
      "|1985|AR47JEX1187B995D81|    1|\n",
      "|   0|AR9AWNF1187B9AB0B4|    1|\n",
      "|1972|ARB29H41187B98F0EF|    1|\n",
      "|1994|ARNF6401187FB57032|    1|\n",
      "|   0|AR10USD1187B99F3F1|    1|\n",
      "|   0|ARKFYS91187B98E58F|    1|\n",
      "|2005|ARDNS031187B9924F0|    1|\n",
      "|   0|ARDR4AC1187FB371A1|    1|\n",
      "|   0|ARLTWXK1187FB5A3F8|    1|\n",
      "|   0|AREBBGV1187FB523D2|    1|\n",
      "|2004|ARMAC4T1187FB3FA4C|    1|\n",
      "|   0|ARQGYP71187FB44566|    1|\n",
      "|1997|AROUOZZ1187B9ABE51|    1|\n",
      "+----+------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_songs.groupBy(['year', 'artist_id']).count().sort(desc('count')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "1ad0ba1e",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_songs = df_songs.repartition('year', 'artist_id')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f46ec56a",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af6dcb34",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "32ed598b",
   "metadata": {},
   "source": [
    "## Artists Table"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "df754e4e",
   "metadata": {},
   "source": [
    "artist_id, name, location, <font color='red'>lattitude</font>, longitude"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c30e863",
   "metadata": {},
   "source": [
    "https://stackoverflow.com/questions/34077353/how-to-change-dataframe-column-names-in-pyspark <br>\n",
    "https://knowledge.udacity.com/questions/678149"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "9dae21a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.03940892219543457\n",
      "71\n",
      "root\n",
      " |-- artist_id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>ARPFHN61187FB575F6</td>\n",
       "      <td>Lupe Fiasco</td>\n",
       "      <td>Chicago, IL</td>\n",
       "      <td>41.88415</td>\n",
       "      <td>-87.63241</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>ARI2JSK1187FB496EF</td>\n",
       "      <td>Nick Ingman;Gavyn Wright</td>\n",
       "      <td>London, England</td>\n",
       "      <td>51.50632</td>\n",
       "      <td>-0.12714</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</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>3</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>4</th>\n",
       "      <td>AR9AWNF1187B9AB0B4</td>\n",
       "      <td>Kenny G featuring Daryl Hall</td>\n",
       "      <td>Seattle, Washington USA</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "            artist_id                          name                 location  \\\n",
       "0  ARPFHN61187FB575F6                   Lupe Fiasco              Chicago, IL   \n",
       "1  ARI2JSK1187FB496EF      Nick Ingman;Gavyn Wright          London, England   \n",
       "2  ARMBR4Y1187B9990EB                  David Martin          California - SF   \n",
       "3  ARPBNLO1187FB3D52F                      Tiny Tim             New York, NY   \n",
       "4  AR9AWNF1187B9AB0B4  Kenny G featuring Daryl Hall  Seattle, Washington USA   \n",
       "\n",
       "   latitude  longitude  \n",
       "0  41.88415  -87.63241  \n",
       "1  51.50632   -0.12714  \n",
       "2  37.77916 -122.42005  \n",
       "3  40.71455  -74.00712  \n",
       "4       NaN        NaN  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# extract columns to create artists table\n",
    "t0 = time()\n",
    "\n",
    "cols = ['artist_name', 'artist_location', 'artist_latitude', 'artist_longitude']\n",
    "cols = [col + ' as ' + col[7:] for col in cols]\n",
    "artists_table = df_songs.selectExpr('artist_id', *cols)\n",
    "\n",
    "print(time()-t0)\n",
    "print(artists_table.count())\n",
    "artists_table.printSchema()\n",
    "artists_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "886f8d54",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # extract columns to create artists table\n",
    "# t0 = time()\n",
    "\n",
    "# columns = ['artist_name', 'artist_location', 'artist_latitude', 'artist_longitude']\n",
    "# columns = [col + ' as ' + col.replace('artist_', '') for col in columns]\n",
    "# artists_table = df.selectExpr('artist_id', *columns)\n",
    "\n",
    "# print(time()-t0)\n",
    "# print(artists_table.count())\n",
    "# artists_table.printSchema()\n",
    "# artists_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "ea4cd9d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # extract columns to create artists table\n",
    "# t0 = time()\n",
    "\n",
    "# artists_table = df.selectExpr([\"artist_id\",\n",
    "#                            \"artist_name as name\", \n",
    "#                            \"artist_location as location\", \n",
    "#                            \"artist_latitude as latitude\",\n",
    "#                            \"artist_longitude as longitude\"])\\\n",
    "#                   .dropDuplicates()\n",
    "\n",
    "# print(time()-t0)\n",
    "# print(artists_table.count())\n",
    "# artists_table.printSchema()\n",
    "# artists_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "38bbc6e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# artists_table.withColumn(\"prefix\", substring('artist_name', 0, 2))\\\n",
    "#              .write.mode(\"overwrite\")\\\n",
    "#              .partitionBy(\"prefix\")\\\n",
    "#              .parquet(\"./artists\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "a8885000",
   "metadata": {},
   "outputs": [],
   "source": [
    "# write artists table to parquet files\n",
    "artists_table.write\\\n",
    ".parquet('./artists', mode='overwrite')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "177166ff",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3da579d9",
   "metadata": {},
   "source": [
    "# Log Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "9f52242d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8056\n",
      "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"
     ]
    },
    {
     "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/5...</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/5...</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/5...</td>\n",
       "      <td>26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>None</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Wyatt</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Scott</td>\n",
       "      <td>NaN</td>\n",
       "      <td>free</td>\n",
       "      <td>Eureka-Arcata-Fortuna, CA</td>\n",
       "      <td>GET</td>\n",
       "      <td>Home</td>\n",
       "      <td>1.540872e+12</td>\n",
       "      <td>563</td>\n",
       "      <td>None</td>\n",
       "      <td>200</td>\n",
       "      <td>1542247071796</td>\n",
       "      <td>Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7....</td>\n",
       "      <td>9</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>None</td>\n",
       "      <td>Logged In</td>\n",
       "      <td>Austin</td>\n",
       "      <td>M</td>\n",
       "      <td>0</td>\n",
       "      <td>Rosales</td>\n",
       "      <td>NaN</td>\n",
       "      <td>free</td>\n",
       "      <td>New York-Newark-Jersey City, NY-NJ-PA</td>\n",
       "      <td>GET</td>\n",
       "      <td>Home</td>\n",
       "      <td>1.541060e+12</td>\n",
       "      <td>521</td>\n",
       "      <td>None</td>\n",
       "      <td>200</td>\n",
       "      <td>1542252577796</td>\n",
       "      <td>Mozilla/5.0 (Windows NT 6.1; rv:31.0) Gecko/20...</td>\n",
       "      <td>12</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        artist       auth firstName gender  itemInSession lastName     length  \\\n",
       "0     Harmonia  Logged In      Ryan      M              0    Smith  655.77751   \n",
       "1  The Prodigy  Logged In      Ryan      M              1    Smith  260.07465   \n",
       "2        Train  Logged In      Ryan      M              2    Smith  205.45261   \n",
       "3         None  Logged In     Wyatt      M              0    Scott        NaN   \n",
       "4         None  Logged In    Austin      M              0  Rosales        NaN   \n",
       "\n",
       "  level                               location method      page  registration  \\\n",
       "0  free     San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong  1.541017e+12   \n",
       "1  free     San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong  1.541017e+12   \n",
       "2  free     San Jose-Sunnyvale-Santa Clara, CA    PUT  NextSong  1.541017e+12   \n",
       "3  free              Eureka-Arcata-Fortuna, CA    GET      Home  1.540872e+12   \n",
       "4  free  New York-Newark-Jersey City, NY-NJ-PA    GET      Home  1.541060e+12   \n",
       "\n",
       "   sessionId             song  status             ts  \\\n",
       "0        583    Sehr kosmisch     200  1542241826796   \n",
       "1        583  The Big Gundown     200  1542242481796   \n",
       "2        583         Marry Me     200  1542242741796   \n",
       "3        563             None     200  1542247071796   \n",
       "4        521             None     200  1542252577796   \n",
       "\n",
       "                                           userAgent userId  \n",
       "0  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...     26  \n",
       "1  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...     26  \n",
       "2  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...     26  \n",
       "3  Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7....      9  \n",
       "4  Mozilla/5.0 (Windows NT 6.1; rv:31.0) Gecko/20...     12  "
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# get filepath to log data file\n",
    "log_data = \"data/log_data/*.json\"\n",
    "\n",
    "# read log data file\n",
    "df_logs = spark.read.json(log_data)\n",
    "\n",
    "print(df_logs.count())\n",
    "df_logs.printSchema()\n",
    "df_logs.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "40e17f01",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6820\n",
      "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"
     ]
    },
    {
     "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/5...</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/5...</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/5...</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...</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...</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                                          userAgent  \\\n",
       "0     200  1542241826796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "1     200  1542242481796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "2     200  1542242741796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "3     200  1542253449796  \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4...   \n",
       "4     200  1542260935796  \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4...   \n",
       "\n",
       "  userId  \n",
       "0     26  \n",
       "1     26  \n",
       "2     26  \n",
       "3     61  \n",
       "4     80  "
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# filter by actions for song plays\n",
    "df_logs = df_logs.filter(\" page = 'NextSong'\")\n",
    "# OR\n",
    "# df_logs = df_logs.filter(df_logs['page'] == \"NextSong\")\n",
    "# OR\n",
    "# df_logs = df_logs.where(df_logs['page'] == \"NextSong\")\n",
    "\n",
    "\n",
    "\n",
    "print(df_logs.count())\n",
    "df_logs.printSchema()\n",
    "df_logs.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d185be11",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76d3042a",
   "metadata": {},
   "source": [
    "## Users Table"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0e4bffd1",
   "metadata": {},
   "source": [
    "user_id, first_name, last_name, gender, level"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "a4594a21",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # extract columns for users table    \n",
    "# users_table = df_logs.select('userId', 'firstName', 'lastName', 'gender', 'level').dropDuplicates()\\\n",
    "# .withColumnRenamed('userId', 'user_id')\\\n",
    "# .withColumnRenamed('lastName', 'last_name')\\\n",
    "# .withColumnRenamed('firstName', 'first_name')\n",
    "\n",
    "# print(users_table.count())\n",
    "# users_table.printSchema()\n",
    "# users_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "0abe8282",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "104\n",
      "root\n",
      " |-- user_id: string (nullable = true)\n",
      " |-- first_name: string (nullable = true)\n",
      " |-- last_name: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "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>26</td>\n",
       "      <td>Ryan</td>\n",
       "      <td>Smith</td>\n",
       "      <td>M</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>81</td>\n",
       "      <td>Sienna</td>\n",
       "      <td>Colon</td>\n",
       "      <td>F</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>75</td>\n",
       "      <td>Joseph</td>\n",
       "      <td>Gutierrez</td>\n",
       "      <td>M</td>\n",
       "      <td>free</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>16</td>\n",
       "      <td>Rylan</td>\n",
       "      <td>George</td>\n",
       "      <td>M</td>\n",
       "      <td>paid</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>54</td>\n",
       "      <td>Kaleb</td>\n",
       "      <td>Cook</td>\n",
       "      <td>M</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      26       Ryan      Smith      M  free\n",
       "1      81     Sienna      Colon      F  free\n",
       "2      75     Joseph  Gutierrez      M  free\n",
       "3      16      Rylan     George      M  paid\n",
       "4      54      Kaleb       Cook      M  free"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# extract columns for users table    \n",
    "users_table = df_logs.selectExpr(\"userId as user_id\", \n",
    "                                 \"firstName as first_name\",\n",
    "                                 \"lastName as last_name\",\n",
    "                                 \"gender\", \n",
    "                                 \"level\").dropDuplicates()\n",
    "\n",
    "print(users_table.count())\n",
    "users_table.printSchema()\n",
    "users_table.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "c7065f07",
   "metadata": {},
   "outputs": [],
   "source": [
    "# extract columns for users table    \n",
    "# users_table = df.selectExpr(['user_id', 'firstName as first_name', 'lastName as last_name', 'gender', 'level', 'ts'])\n",
    "# users_window = Window.partitionBy('user_id').orderBy(F.desc('ts'))\n",
    "# users_table = users_table.withColumn('row_number', F.row_number().over(users_window))\n",
    "# users_table = users_table.where(users_table.row_number == 1).drop('ts', 'row_number')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "10dcfff1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# write users table to parquet files\n",
    "users_table.write.parquet(\"./users\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d3c686aa",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5a4b835",
   "metadata": {},
   "source": [
    "## Time Table\n",
    "start_time, hour, day, week, month, year, weekday"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "36556727",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6820\n",
      "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",
      " |-- start_time: timestamp (nullable = true)\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "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>start_time</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/5...</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 02: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/5...</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 02: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/5...</td>\n",
       "      <td>26</td>\n",
       "      <td>2018-11-15 02: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...</td>\n",
       "      <td>61</td>\n",
       "      <td>2018-11-15 05: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...</td>\n",
       "      <td>80</td>\n",
       "      <td>2018-11-15 07: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                                          userAgent  \\\n",
       "0     200  1542241826796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "1     200  1542242481796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "2     200  1542242741796  \"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/5...   \n",
       "3     200  1542253449796  \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4...   \n",
       "4     200  1542260935796  \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4...   \n",
       "\n",
       "  userId              start_time  \n",
       "0     26 2018-11-15 02:30:26.796  \n",
       "1     26 2018-11-15 02:41:21.796  \n",
       "2     26 2018-11-15 02:45:41.796  \n",
       "3     61 2018-11-15 05:44:09.796  \n",
       "4     80 2018-11-15 07:48:55.796  "
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create timestamp column from original timestamp column\n",
    "get_timestamp = udf(lambda ts: datetime.fromtimestamp(ts/1000), TimestampType())\n",
    "df_logs = df_logs.withColumn('start_time', get_timestamp('ts'))\n",
    "    \n",
    "# # create datetime column from original timestamp column\n",
    "# get_datetime = udf()\n",
    "# df_logs = \n",
    "\n",
    "print(df_logs.count())\n",
    "df_logs.printSchema()\n",
    "df_logs.limit(5).toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "c012325a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # create timestamp column from original timestamp column\n",
    "# get_timestamp = udf(lambda ts: datetime.fromtimestamp(ts/1000).isoformat())\n",
    "# df_logs = df_logs.withColumn('start_time', get_timestamp('ts').cast(TimestampType()))\n",
    "    \n",
    "# # # create datetime column from original timestamp column\n",
    "# # get_datetime = udf()\n",
    "# # df_logs = "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "d993b5bf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "6820\n",
      "root\n",
      " |-- start_time: timestamp (nullable = true)\n",
      " |-- hour: integer (nullable = true)\n",
      " |-- day: integer (nullable = true)\n",
      " |-- week: integer (nullable = true)\n",
      " |-- month: integer (nullable = true)\n",
      " |-- year: integer (nullable = true)\n",
      " |-- weekday: integer (nullable = true)\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "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 02:30:26.796</td>\n",
       "      <td>2</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 02:41:21.796</td>\n",
       "      <td>2</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 02:45:41.796</td>\n",
       "      <td>2</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 05:44:09.796</td>\n",
       "      <td>5</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-15 07:48:55.796</td>\n",
       "      <td>7</td>\n",
       "      <td>15</td>\n",
       "      <td>46</td>\n",
       "      <td>11</td>\n",
       "      <td>2018</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               start_time  hour  day  week  month  year  weekday\n",
       "0 2018-11-15 02:30:26.796     2   15    46     11  2018        5\n",
       "1 2018-11-15 02:41:21.796     2   15    46     11  2018        5\n",
       "2 2018-11-15 02:45:41.796     2   15    46     11  2018        5\n",
       "3 2018-11-15 05:44:09.796     5   15    46     11  2018        5\n",
       "4 2018-11-15 07:48:55.796     7   15    46     11  2018        5"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# extract columns to create time table\n",
    "time_table = df_logs.select('start_time')\n",
    "# list of functions\n",
    "funcs = [F.hour, F.dayofmonth, F.weekofyear, F.month, F.year, F.dayofweek]\n",
    "cols = ['hour', 'day', 'week', 'month', 'year', 'weekday']\n",
    "time_cols = [(col, func('start_time'))for func, col in zip(funcs, cols)]\n",
    "for col in time_cols:\n",
    "    time_table = time_table.withColumn(*col)\n",
    "    \n",
    "print(time_table.count())\n",
    "time_table.printSchema()\n",
    "time_table.limit(5).toPandas() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "63ea7df7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # extract columns to create time table\n",
    "# time_table = df_logs.select('start_time')\n",
    "# time_table = time_table.withColumn('hour', F.hour('start_time'))\\\n",
    "#                        .withColumn('day', F.dayofmonth('start_time'))\\\n",
    "#                        .withColumn('week', F.weekofyear('start_time'))\\\n",
    "#                        .withColumn('month', F.month('start_time'))\\\n",
    "#                        .withColumn('year', F.year('start_time'))\\\n",
    "#                        .withColumn('weekday', F.dayofweek('start_time'))\n",
    "\n",
    "# print(time_table.count())\n",
    "# time_table.printSchema()\n",
    "# time_table.limit(5).toPandas() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "c054b866",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # extract columns to create time table\n",
    "# time_table = df_logs.select('start_time')\n",
    "# time_table = time_table.withColumn('hour', F.hour('start_time'))\n",
    "# time_table = time_table.withColumn('day', F.dayofmonth('start_time'))\n",
    "# time_table = time_table.withColumn('week', F.weekofyear('start_time'))\n",
    "# time_table = time_table.withColumn('month', F.month('start_time'))\n",
    "# time_table = time_table.withColumn('year', F.year('start_time'))\n",
    "# time_table = time_table.withColumn('weekday', F.dayofweek('start_time'))\n",
    "\n",
    "# print(time_table.count())\n",
    "# time_table.printSchema()\n",
    "# time_table.limit(5).toPandas() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "2c0fa983",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# write time table to parquet files partitioned by year and month\n",
    "time_table.write.partitionBy(\"year\", \"month\").parquet(\"times\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8da9e395",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1c93297a",
   "metadata": {},
   "source": [
    "## SongPlay Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "f1480618",
   "metadata": {},
   "outputs": [],
   "source": [
    "# # read in song data to use for songplays table\n",
    "# song_df = spark.read.parquet(\"songs/*/*/*.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "344d75de",
   "metadata": {},
   "outputs": [],
   "source": [
    "# extract columns from joined song and log datasets to create songplays table\n",
    "df_logs = df_logs.orderBy('ts')\n",
    "df_logs = df_logs.withColumn('songplay_id', F.monotonically_increasing_id())\n",
    "\n",
    "df_songs.createOrReplaceTempView('songs')\n",
    "df_logs.createOrReplaceTempView('events')\n",
    "\n",
    "# include year and month to allow parquet partitioning\n",
    "songplays_table = spark.sql(\"\"\"\n",
    "        SELECT\n",
    "            e.songplay_id,\n",
    "            e.start_time,\n",
    "            e.userId as user_id,\n",
    "            e.level,\n",
    "            s.song_id,\n",
    "            s.artist_id,\n",
    "            e.sessionId as session_id,\n",
    "            e.location,\n",
    "            e.userAgent as user_agent,\n",
    "            year(e.start_time) as year,\n",
    "            month(e.start_time) as month\n",
    "        FROM events e\n",
    "        LEFT JOIN songs s ON\n",
    "            e.song = s.title AND\n",
    "            e.artist = s.artist_name AND\n",
    "            ABS(e.length - s.duration) < 2\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "3442e7f9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# write songplays table to parquet files partitioned by year and month\n",
    "songplays_table.write.parquet('./songplays', \n",
    "                              partitionBy=['year', 'month'],\n",
    "                              mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a604fa0a",
   "metadata": {},
   "source": [
    "**A quick question to brainstorm here : how can you partition the data without including the partition columns in the dataset?** \n",
    "- the partition colums are not included in the parquet files, the arey only used for naming the files."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0bba6f71",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ef7fbde8",
   "metadata": {},
   "source": [
    "### RUN PYTHON SCRIPTS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "436e9621",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/10/12 04:14:29 WARN Utils: Your hostname, Mahmouds-MacBook-Pro.local resolves to a loopback address: 127.0.0.1; using 192.168.1.2 instead (on interface en0)\n",
      "22/10/12 04:14:29 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      ":: loading settings :: url = jar:file:/Users/mnagy99/opt/anaconda3/lib/python3.8/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n",
      "Ivy Default Cache set to: /Users/mnagy99/.ivy2/cache\n",
      "The jars for the packages stored in: /Users/mnagy99/.ivy2/jars\n",
      "org.apache.hadoop#hadoop-aws added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-66559092-4991-4002-94c4-1fd5cc3edb29;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.hadoop#hadoop-aws;2.7.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-common;2.7.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-annotations;2.7.0 in central\n",
      "\tfound com.google.guava#guava;11.0.2 in central\n",
      "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n",
      "\tfound commons-cli#commons-cli;1.2 in central\n",
      "\tfound org.apache.commons#commons-math3;3.1.1 in central\n",
      "\tfound xmlenc#xmlenc;0.52 in central\n",
      "\tfound commons-httpclient#commons-httpclient;3.1 in central\n",
      "\tfound commons-logging#commons-logging;1.1.3 in central\n",
      "\tfound commons-codec#commons-codec;1.4 in central\n",
      "\tfound commons-io#commons-io;2.4 in central\n",
      "\tfound commons-net#commons-net;3.1 in central\n",
      "\tfound commons-collections#commons-collections;3.2.1 in central\n",
      "\tfound javax.servlet#servlet-api;2.5 in central\n",
      "\tfound org.mortbay.jetty#jetty;6.1.26 in central\n",
      "\tfound org.mortbay.jetty#jetty-util;6.1.26 in central\n",
      "\tfound com.sun.jersey#jersey-core;1.9 in central\n",
      "\tfound com.sun.jersey#jersey-json;1.9 in central\n",
      "\tfound org.codehaus.jettison#jettison;1.1 in central\n",
      "\tfound com.sun.xml.bind#jaxb-impl;2.2.3-1 in central\n",
      "\tfound javax.xml.bind#jaxb-api;2.2.2 in central\n",
      "\tfound javax.xml.stream#stax-api;1.0-2 in central\n",
      "\tfound javax.activation#activation;1.1 in central\n",
      "\tfound org.codehaus.jackson#jackson-core-asl;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-mapper-asl;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-jaxrs;1.9.13 in central\n",
      "\tfound org.codehaus.jackson#jackson-xc;1.9.13 in central\n",
      "\tfound com.sun.jersey#jersey-server;1.9 in central\n",
      "\tfound asm#asm;3.2 in central\n",
      "\tfound log4j#log4j;1.2.17 in central\n",
      "\tfound net.java.dev.jets3t#jets3t;0.9.0 in central\n",
      "\tfound org.apache.httpcomponents#httpclient;4.2.5 in central\n",
      "\tfound org.apache.httpcomponents#httpcore;4.2.5 in central\n",
      "\tfound com.jamesmurty.utils#java-xmlbuilder;0.4 in central\n",
      "\tfound commons-lang#commons-lang;2.6 in central\n",
      "\tfound commons-configuration#commons-configuration;1.6 in central\n",
      "\tfound commons-digester#commons-digester;1.8 in central\n",
      "\tfound commons-beanutils#commons-beanutils;1.7.0 in central\n",
      "\tfound commons-beanutils#commons-beanutils-core;1.8.0 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.10 in central\n",
      "\tfound org.apache.avro#avro;1.7.4 in central\n",
      "\tfound com.thoughtworks.paranamer#paranamer;2.3 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.0.4.1 in central\n",
      "\tfound org.apache.commons#commons-compress;1.4.1 in central\n",
      "\tfound org.tukaani#xz;1.0 in central\n",
      "\tfound com.google.protobuf#protobuf-java;2.5.0 in central\n",
      "\tfound com.google.code.gson#gson;2.2.4 in central\n",
      "\tfound org.apache.hadoop#hadoop-auth;2.7.0 in central\n",
      "\tfound org.apache.directory.server#apacheds-kerberos-codec;2.0.0-M15 in central\n",
      "\tfound org.apache.directory.server#apacheds-i18n;2.0.0-M15 in central\n",
      "\tfound org.apache.directory.api#api-asn1-api;1.0.0-M20 in central\n",
      "\tfound org.apache.directory.api#api-util;1.0.0-M20 in central\n",
      "\tfound org.apache.zookeeper#zookeeper;3.4.6 in central\n",
      "\tfound org.slf4j#slf4j-log4j12;1.7.10 in central\n",
      "\tfound io.netty#netty;3.6.2.Final in central\n",
      "\tfound org.apache.curator#curator-framework;2.7.1 in central\n",
      "\tfound org.apache.curator#curator-client;2.7.1 in central\n",
      "\tfound com.jcraft#jsch;0.1.42 in central\n",
      "\tfound org.apache.curator#curator-recipes;2.7.1 in central\n",
      "\tfound org.apache.htrace#htrace-core;3.1.0-incubating in central\n",
      "\tfound javax.servlet.jsp#jsp-api;2.1 in central\n",
      "\tfound jline#jline;0.9.94 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-databind;2.2.3 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-annotations;2.2.3 in central\n",
      "\tfound com.fasterxml.jackson.core#jackson-core;2.2.3 in central\n",
      "\tfound com.amazonaws#aws-java-sdk;1.7.4 in central\n",
      "\tfound joda-time#joda-time;2.11.2 in central\n",
      "\t[2.11.2] joda-time#joda-time;[2.2,)\n",
      ":: resolution report :: resolve 4884ms :: artifacts dl 86ms\n",
      "\t:: modules in use:\n",
      "\tasm#asm;3.2 from central in [default]\n",
      "\tcom.amazonaws#aws-java-sdk;1.7.4 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-annotations;2.2.3 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-core;2.2.3 from central in [default]\n",
      "\tcom.fasterxml.jackson.core#jackson-databind;2.2.3 from central in [default]\n",
      "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n",
      "\tcom.google.code.gson#gson;2.2.4 from central in [default]\n",
      "\tcom.google.guava#guava;11.0.2 from central in [default]\n",
      "\tcom.google.protobuf#protobuf-java;2.5.0 from central in [default]\n",
      "\tcom.jamesmurty.utils#java-xmlbuilder;0.4 from central in [default]\n",
      "\tcom.jcraft#jsch;0.1.42 from central in [default]\n",
      "\tcom.sun.jersey#jersey-core;1.9 from central in [default]\n",
      "\tcom.sun.jersey#jersey-json;1.9 from central in [default]\n",
      "\tcom.sun.jersey#jersey-server;1.9 from central in [default]\n",
      "\tcom.sun.xml.bind#jaxb-impl;2.2.3-1 from central in [default]\n",
      "\tcom.thoughtworks.paranamer#paranamer;2.3 from central in [default]\n",
      "\tcommons-beanutils#commons-beanutils;1.7.0 from central in [default]\n",
      "\tcommons-beanutils#commons-beanutils-core;1.8.0 from central in [default]\n",
      "\tcommons-cli#commons-cli;1.2 from central in [default]\n",
      "\tcommons-codec#commons-codec;1.4 from central in [default]\n",
      "\tcommons-collections#commons-collections;3.2.1 from central in [default]\n",
      "\tcommons-configuration#commons-configuration;1.6 from central in [default]\n",
      "\tcommons-digester#commons-digester;1.8 from central in [default]\n",
      "\tcommons-httpclient#commons-httpclient;3.1 from central in [default]\n",
      "\tcommons-io#commons-io;2.4 from central in [default]\n",
      "\tcommons-lang#commons-lang;2.6 from central in [default]\n",
      "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n",
      "\tcommons-net#commons-net;3.1 from central in [default]\n",
      "\tio.netty#netty;3.6.2.Final from central in [default]\n",
      "\tjavax.activation#activation;1.1 from central in [default]\n",
      "\tjavax.servlet#servlet-api;2.5 from central in [default]\n",
      "\tjavax.servlet.jsp#jsp-api;2.1 from central in [default]\n",
      "\tjavax.xml.bind#jaxb-api;2.2.2 from central in [default]\n",
      "\tjavax.xml.stream#stax-api;1.0-2 from central in [default]\n",
      "\tjline#jline;0.9.94 from central in [default]\n",
      "\tjoda-time#joda-time;2.11.2 from central in [default]\n",
      "\tlog4j#log4j;1.2.17 from central in [default]\n",
      "\tnet.java.dev.jets3t#jets3t;0.9.0 from central in [default]\n",
      "\torg.apache.avro#avro;1.7.4 from central in [default]\n",
      "\torg.apache.commons#commons-compress;1.4.1 from central in [default]\n",
      "\torg.apache.commons#commons-math3;3.1.1 from central in [default]\n",
      "\torg.apache.curator#curator-client;2.7.1 from central in [default]\n",
      "\torg.apache.curator#curator-framework;2.7.1 from central in [default]\n",
      "\torg.apache.curator#curator-recipes;2.7.1 from central in [default]\n",
      "\torg.apache.directory.api#api-asn1-api;1.0.0-M20 from central in [default]\n",
      "\torg.apache.directory.api#api-util;1.0.0-M20 from central in [default]\n",
      "\torg.apache.directory.server#apacheds-i18n;2.0.0-M15 from central in [default]\n",
      "\torg.apache.directory.server#apacheds-kerberos-codec;2.0.0-M15 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-annotations;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-auth;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-aws;2.7.0 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-common;2.7.0 from central in [default]\n",
      "\torg.apache.htrace#htrace-core;3.1.0-incubating from central in [default]\n",
      "\torg.apache.httpcomponents#httpclient;4.2.5 from central in [default]\n",
      "\torg.apache.httpcomponents#httpcore;4.2.5 from central in [default]\n",
      "\torg.apache.zookeeper#zookeeper;3.4.6 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-core-asl;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-jaxrs;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-mapper-asl;1.9.13 from central in [default]\n",
      "\torg.codehaus.jackson#jackson-xc;1.9.13 from central in [default]\n",
      "\torg.codehaus.jettison#jettison;1.1 from central in [default]\n",
      "\torg.mortbay.jetty#jetty;6.1.26 from central in [default]\n",
      "\torg.mortbay.jetty#jetty-util;6.1.26 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.10 from central in [default]\n",
      "\torg.slf4j#slf4j-log4j12;1.7.10 from central in [default]\n",
      "\torg.tukaani#xz;1.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.0.4.1 from central in [default]\n",
      "\txmlenc#xmlenc;0.52 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   68  |   1   |   0   |   0   ||   68  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-66559092-4991-4002-94c4-1fd5cc3edb29\n",
      "\tconfs: [default]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\t0 artifacts copied, 68 already retrieved (0kB/48ms)\n",
      "22/10/12 04:14:35 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "22/10/12 04:14:36 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# ### How to run the Python scripts\n",
    "# We can run python scripts either:\n",
    "# - in the notebook:\n",
    "#     - %run -i '<filename>.py'\n",
    "#     - !python <filename>.py\n",
    "\n",
    "!python etl.py \n",
    "\n",
    "### (Using S3 with Local Mode!!)\n",
    "### This is not recommended at all in production\n",
    "### what we are doing now is loading the data all the way from amazon,\n",
    "### and into the memory of jupyter Notebook Machine"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b045e90b",
   "metadata": {},
   "source": [
    "- pyspark.sql.utils.AnalysisException: Path does not exist: file:/Users/mnagy99/Desktop/DataLake Project/datasong_data/A/A/A/*.json\n",
    "- pyspark.sql.utils.AnalysisException: Path does not exist: file:/Users/mnagy99/Desktop/DataLake Project/data/log_data/A/A/A.json"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e16d5b9",
   "metadata": {},
   "source": [
    "<br>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b08ab29",
   "metadata": {},
   "source": [
    "<br>"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}