Data Wrangling with Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Spark with Jupyter Notebook on MacOS (2.0.0 and higher)](https://medium.com/@roshinijohri/spark-with-jupyter-notebook-on-macos-2-0-0-and-higher-c61b971b5007)\n",
    "==========================================================================================\n",
    "\n",
    "#### Run in Terminal:\n",
    "$\\textrm{brew install apache-spark}$\n",
    "\n",
    "$\\textrm{brew info apache-spark}$\n",
    "\n",
    "$\\textrm{export SPARK_HOME='/usr/local/Cellar/apache-spark/2.4.5/libexec/'}$ -> Edit depending on version\n",
    "\n",
    "$\\textrm{pyspark}$"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\n",
      "      ____              __\n",
      "     / __/__  ___ _____/ /__\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 2.4.5\n",
      "      /_/\n",
      "\n",
      "Using Python version 3.7.4 (default, Aug 13 2019 15:17:50)\n",
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "exec(open(os.path.join(os.environ['SPARK_HOME'], 'python/pyspark/shell.py')).read())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql.session import SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "    .appName('spark test') \\\n",
    "    .getOrCreate() \\\n",
    "\n",
    "columns = ['id', 'dogs', 'cats']\n",
    "vals = [\n",
    "    (1, 2, 0),\n",
    "    (2, 0, 1)\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+----+\n",
      "| id|dogs|cats|\n",
      "+---+----+----+\n",
      "|  1|   2|   0|\n",
      "|  2|   0|   1|\n",
      "+---+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create DataFrame\n",
    "df = spark.createDataFrame(vals, columns)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Maps\n",
    "\n",
    "In Spark, maps take data as input and then transform that data with whatever function you put in the map. They are like directions for the data telling how each input should get to the output.\n",
    "\n",
    "The first code cell creates a SparkContext object. With the SparkContext, you can input a dataset and parallelize the data across a cluster (since you are currently using Spark in local mode on a single machine, technically the dataset isn't distributed yet).\n",
    "\n",
    "Run the code cell below to instantiate a SparkContext object and then read in the `log_of_songs` list into Spark. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The findspark Python module makes it easier to install Spark in local mode on your computer\n",
    "# This is convenient for practicing Spark syntax locally.\n",
    "import findspark\n",
    "findspark.init('/usr/local/Cellar/apache-spark/2.4.5/libexec')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.0.19:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.5</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>maps_and_lazy_evaluation_example</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=maps_and_lazy_evaluation_example>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "\n",
    "# Shut down the SparkContext\n",
    "sc.stop()\n",
    "\n",
    "# Main entry point for Spark functionality; A SparkContext represents the connection to a Spark cluster, \n",
    "#   and can be used to create L{RDD} and broadcast variables on that cluster\n",
    "# Note: master: Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4])\n",
    "#      appName: a name for your job, to display on the cluster web UI\n",
    "sc = SparkContext(appName='maps_and_lazy_evaluation_example')\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Control our logLevel. This overrides any user-defined log settings.\n",
    "#   Valid log levels include: ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, WARN\n",
    "sc.setLogLevel('INFO')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Parallelize the log_of_songs to use with Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "log_of_songs = [\n",
    "    'Despacito',\n",
    "    'Nice for what',\n",
    "    'No tears left to cry',\n",
    "    'Despacito',\n",
    "    'Havana',\n",
    "    'In my feelings',\n",
    "    'Nice for what',\n",
    "    'despacito',\n",
    "    'All the stars'\n",
    "]\n",
    "\n",
    "# Distribute a local Python collection to form an RDD\n",
    "# Using xrange is recommended if the input represents a range for performance\n",
    "distributed_song_log = sc.parallelize(log_of_songs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'havana'"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Define a function that converts a song title to lowercase\n",
    "def convert_song_to_lowercase(song):\n",
    "    return song.lower()\n",
    "\n",
    "convert_song_to_lowercase('Havana')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code cells demonstrate how to apply this function using a map step. The map step will go through each song in the list and apply the `convert_song_to_lowercase()` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[1] at RDD at PythonRDD.scala:53"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Return a new RDD by applying a function to each element of this RDD\n",
    "distributed_song_log.map(convert_song_to_lowercase)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You'll notice that this code cell ran quite quickly. This is because of lazy evaluation. Spark does not actually execute the map step unless it needs to.\n",
    "\n",
    "\"RDD\" in the output refers to resilient distributed dataset. RDDs are exactly what they say they are: fault-tolerant datasets distributed across a cluster. This is how Spark stores data. \n",
    "\n",
    "To get Spark to actually run the map step, you need to use an \"action\". One available action is the collect method. The `collect()` method takes the results from all of the clusters and \"collects\" them into a single list on the master node."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['despacito',\n",
       " 'nice for what',\n",
       " 'no tears left to cry',\n",
       " 'despacito',\n",
       " 'havana',\n",
       " 'in my feelings',\n",
       " 'nice for what',\n",
       " 'despacito',\n",
       " 'all the stars']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Return a new RDD by applying a function to each element of this RDD\n",
    "# Return a list that contains all of the elements in this RDD\n",
    "# Note: This method should only be used if the resulting array is expected to be small, as all the data \n",
    "#       is loaded into the driver’s memory\n",
    "distributed_song_log.map(convert_song_to_lowercase).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that Spark is not changing the original data set: Spark is merely making a copy. You can see this by running `collect()` on the original dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Despacito',\n",
       " 'Nice for what',\n",
       " 'No tears left to cry',\n",
       " 'Despacito',\n",
       " 'Havana',\n",
       " 'In my feelings',\n",
       " 'Nice for what',\n",
       " 'despacito',\n",
       " 'All the stars']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Return a list that contains all of the elements in this RDD\n",
    "# Note: This method should only be used if the resulting array is expected to be small, as all the data \n",
    "#       is loaded into the driver’s memory\n",
    "distributed_song_log.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You do not always have to write a custom function for the map step. You can also use anonymous (lambda) functions as well as built-in Python functions like `string.lower()`. \n",
    "\n",
    "Anonymous functions are actually a Python feature for writing functional style programs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['despacito',\n",
       " 'nice for what',\n",
       " 'no tears left to cry',\n",
       " 'despacito',\n",
       " 'havana',\n",
       " 'in my feelings',\n",
       " 'nice for what',\n",
       " 'despacito',\n",
       " 'all the stars']"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Return a new RDD by applying a function to each element of this RDD\n",
    "# Return a list that contains all of the elements in this RDD\n",
    "# Note: This method should only be used if the resulting array is expected to be small, as all the data \n",
    "#       is loaded into the driver’s memory\n",
    "distributed_song_log.map(lambda song: song.lower()).collect()\n",
    "#distributed_song_log.map(lambda x: x.lower()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reading and Writing Data with Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since we're using Spark locally we already have both a sparkcontext and a sparksession running. We can update some of the parameters, such our application's name. Let's just call it \"Our first Python Spark SQL example\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instantiate a Spark session \n",
    "# The entry point to programming Spark with the Dataset and DataFrame API\n",
    "# Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "#       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "spark = SparkSession.builder \\\n",
    "    .appName('Our first Python Spark SQL example') \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's check if the change went through."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('spark.app.id', 'local-1592506310635'),\n",
       " ('spark.app.name', 'Our first Python Spark SQL example'),\n",
       " ('spark.sql.catalogImplementation', 'hive'),\n",
       " ('spark.rdd.compress', 'True'),\n",
       " ('spark.driver.host', '192.168.0.19'),\n",
       " ('spark.serializer.objectStreamReset', '100'),\n",
       " ('spark.master', 'local[*]'),\n",
       " ('spark.executor.id', 'driver'),\n",
       " ('spark.submit.deployMode', 'client'),\n",
       " ('spark.driver.port', '52013'),\n",
       " ('spark.ui.showConsoleProgress', 'true')]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Get all values as a list of key-value pairs\n",
    "spark.sparkContext.getConf().getAll()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.0.19:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.5</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>maps_and_lazy_evaluation_example</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fac2193f650>"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see the app name is exactly how we set it\n",
    "\n",
    "Let's create our first dataframe from a fairly small sample data set. Througout the course we'll work with a log file data set that describes user interactions with a music streaming service. The records describe events such as logging in to the site, visiting a page, listening to the next song, seeing an ad."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Data Wrangling with Spark/data/sparkify_log_small.json'\n",
    "\n",
    "# Loads JSON files and returns the results as a :class:`DataFrame`\n",
    "# Note: path: string represents path to the JSON dataset, or a list of paths, or RDD of Strings storing JSON objects\n",
    "user_log = spark.read.json(path=path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: long (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prints out the schema in the tree format\n",
    "user_log.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "|summary|           artist|      auth|firstName|gender|     itemInSession|lastName|           length|level|    location|method|   page|        registration|         sessionId|    song|           status|                 ts|           userAgent|            userId|\n",
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "|  count|             8347|     10000|     9664|  9664|             10000|    9664|             8347|10000|        9664| 10000|  10000|                9664|             10000|    8347|            10000|              10000|                9664|             10000|\n",
      "|   mean|            461.0|      null|     null|  null|           19.6734|    null|249.6486587492506| null|        null|  null|   null|1.504695369588739...|         4436.7511|Infinity|         202.8984| 1.5137859954164E12|                null|1442.4413286423842|\n",
      "| stddev|            300.0|      null|     null|  null|25.382114916132608|    null|95.00437130781461| null|        null|  null|   null|  8.47314252131656E9|2043.1281541827561|     NaN|18.04179115450588|3.290828862357974E7|                null| 829.8909432082621|\n",
      "|    min|              !!!|     Guest|   Aakash|     F|                 0| Acevedo|          1.12281| free|Aberdeen, WA|   GET|  About|       1463503881284|                 9|      #1|              200|      1513720872284|\"Mozilla/5.0 (Mac...|                  |\n",
      "|    max|ÃÂlafur Arnalds|Logged Out|     Zoie|     M|               163|  Zuniga|        1806.8371| paid|    Yuma, AZ|   PUT|Upgrade|       1513760702284|              7144|wingless|              404|      1513848349284|Mozilla/5.0 (comp...|               999|\n",
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Computes basic statistics for numeric and string columns, this include count, mean, stddev, min, and max\n",
    "#   If no columns are given, this function computes statistics for all numerical or string columns\n",
    "user_log.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+\n",
      "|summary|           length|\n",
      "+-------+-----------------+\n",
      "|  count|             8347|\n",
      "|   mean|249.6486587492506|\n",
      "| stddev|95.00437130781461|\n",
      "|    min|          1.12281|\n",
      "|    max|        1806.8371|\n",
      "+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Computes basic statistics for numeric and string columns, this include count, mean, stddev, min, and max\n",
    "#   If no columns are given, this function computes statistics for all numerical or string columns\n",
    "user_log.describe('length').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|       artist|     auth|firstName|gender|itemInSession|lastName|   length|level|            location|method|    page| registration|sessionId|                song|status|           ts|           userAgent|userId|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|Showaddywaddy|Logged In|  Kenneth|     M|          112|Matthews|232.93342| paid|Charlotte-Concord...|   PUT|NextSong|1509380319284|     5132|Christmas Tears W...|   200|1513720872284|\"Mozilla/5.0 (Win...|  1046|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prints the first n rows to the console\n",
    "# Note: n: Number of rows to show\n",
    "#       truncate: whether truncate long strings and align cells right\n",
    "user_log.show(n=1, truncate=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046'),\n",
       " Row(artist='Lily Allen', auth='Logged In', firstName='Elizabeth', gender='F', itemInSession=7, lastName='Chase', length=195.23873, level='free', location='Shreveport-Bossier City, LA', method='PUT', page='NextSong', registration=1512718541284, sessionId=5027, song='Cheryl Tweedy', status=200, ts=1513720878284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='1000'),\n",
       " Row(artist='Cobra Starship Featuring Leighton Meester', auth='Logged In', firstName='Vera', gender='F', itemInSession=6, lastName='Blackwell', length=196.20526, level='paid', location='Racine, WI', method='PUT', page='NextSong', registration=1499855749284, sessionId=5516, song='Good Girls Go Bad (Feat.Leighton Meester) (Album Version)', status=200, ts=1513720881284, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.78.2 (KHTML, like Gecko) Version/7.0.6 Safari/537.78.2\"', userId='2219'),\n",
       " Row(artist='Alex Smoke', auth='Logged In', firstName='Sophee', gender='F', itemInSession=8, lastName='Barker', length=405.99465, level='paid', location='San Luis Obispo-Paso Robles-Arroyo Grande, CA', method='PUT', page='NextSong', registration=1513009647284, sessionId=2372, song=\"Don't See The Point\", status=200, ts=1513720905284, userAgent='\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='2373'),\n",
       " Row(artist=None, auth='Logged In', firstName='Jordyn', gender='F', itemInSession=0, lastName='Jones', length=None, level='free', location='Syracuse, NY', method='GET', page='Home', registration=1513648531284, sessionId=1746, song=None, status=200, ts=1513720913284, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.94 Safari/537.36\"', userId='1747')]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``num`` rows as a :class:`list` of :class:`Row`\n",
    "user_log.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "out_path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Data Wrangling with Spark/data/sparkify_log_small.csv'\n",
    "\n",
    "# Saves the contents of the :class:`DataFrame` to a data source\n",
    "# Note: path: the path in a Hadoop supported file system\n",
    "#       format: the format used to save\n",
    "#       mode: specifies the behavior of the save operation when data already exists\n",
    "#             * ``append``: Append contents of this :class:`DataFrame` to existing data.\n",
    "#             * ``overwrite``: Overwrite existing data.\n",
    "#             * ``ignore``: Silently ignore this operation if data already exists.\n",
    "#             * ``error`` or ``errorifexists`` (default case): Throw an exception if data already exists\n",
    "#       header: uses the first line as names of columns; if None is set, it uses the default value, ``false``\n",
    "user_log.write.save(path=out_path, format='csv', mode='overwrite', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Loads a CSV file and returns the result as a :class:`DataFrame`\n",
    "# path: string, or list of strings, for input path(s), or RDD of Strings storing CSV rows\n",
    "# header: uses the first line as names of columns; if None is set, it uses the default value, ``false``\n",
    "user_log_2 = spark.read.csv(path=out_path, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: string (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: string (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: string (nullable = true)\n",
      " |-- sessionId: string (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- ts: string (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prints out the schema in the tree format\n",
    "user_log_2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession='112', lastName='Matthews', length='232.93342', level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration='1509380319284', sessionId='5132', song='Christmas Tears Will Fall', status='200', ts='1513720872284', userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046'),\n",
       " Row(artist='Lily Allen', auth='Logged In', firstName='Elizabeth', gender='F', itemInSession='7', lastName='Chase', length='195.23873', level='free', location='Shreveport-Bossier City, LA', method='PUT', page='NextSong', registration='1512718541284', sessionId='5027', song='Cheryl Tweedy', status='200', ts='1513720878284', userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='1000')]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``num`` rows as a :class:`list` of :class:`Row`\n",
    "user_log_2.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userID|\n",
      "+------+\n",
      "|  1046|\n",
      "|  1000|\n",
      "|  2219|\n",
      "|  2373|\n",
      "|  1747|\n",
      "|  1747|\n",
      "|  1162|\n",
      "|  1061|\n",
      "|   748|\n",
      "|   597|\n",
      "|  1806|\n",
      "|   748|\n",
      "|  1176|\n",
      "|  2164|\n",
      "|  2146|\n",
      "|  2219|\n",
      "|  1176|\n",
      "|  2904|\n",
      "|   597|\n",
      "|   226|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "user_log_2.select('userID').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession='112', lastName='Matthews', length='232.93342', level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration='1509380319284', sessionId='5132', song='Christmas Tears Will Fall', status='200', ts='1513720872284', userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046')]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``num`` rows as a :class:`list` of :class:`Row`\n",
    "user_log_2.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Wrangling with Spark DataFrames\n",
    "\n",
    "Helpful resources:\n",
    "http://spark.apache.org/docs/latest/api/python/pyspark.sql.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.functions import isnan, count, when, col, sort_array, avg\n",
    "from pyspark.sql.functions import asc, desc\n",
    "from pyspark.sql.functions import sum as Fsum\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "import datetime\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instantiate a Spark session \n",
    "# The entry point to programming Spark with the Dataset and DataFrame API\n",
    "# Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "#       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "spark = SparkSession.builder \\\n",
    "    .appName('Data Wrangling with Spark DataFrame') \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Data Wrangling with Spark/data/sparkify_log_small.json'\n",
    "\n",
    "# Loads JSON files and returns the results as a :class:`DataFrame`\n",
    "# Note: path: string represents path to the JSON dataset, or a list of paths, or RDD of Strings storing JSON objects\n",
    "user_log = spark.read.json(path=path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data Exploration "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: long (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prints out the schema in the tree format\n",
    "user_log.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046'),\n",
       " Row(artist='Lily Allen', auth='Logged In', firstName='Elizabeth', gender='F', itemInSession=7, lastName='Chase', length=195.23873, level='free', location='Shreveport-Bossier City, LA', method='PUT', page='NextSong', registration=1512718541284, sessionId=5027, song='Cheryl Tweedy', status=200, ts=1513720878284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='1000'),\n",
       " Row(artist='Cobra Starship Featuring Leighton Meester', auth='Logged In', firstName='Vera', gender='F', itemInSession=6, lastName='Blackwell', length=196.20526, level='paid', location='Racine, WI', method='PUT', page='NextSong', registration=1499855749284, sessionId=5516, song='Good Girls Go Bad (Feat.Leighton Meester) (Album Version)', status=200, ts=1513720881284, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.78.2 (KHTML, like Gecko) Version/7.0.6 Safari/537.78.2\"', userId='2219'),\n",
       " Row(artist='Alex Smoke', auth='Logged In', firstName='Sophee', gender='F', itemInSession=8, lastName='Barker', length=405.99465, level='paid', location='San Luis Obispo-Paso Robles-Arroyo Grande, CA', method='PUT', page='NextSong', registration=1513009647284, sessionId=2372, song=\"Don't See The Point\", status=200, ts=1513720905284, userAgent='\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='2373'),\n",
       " Row(artist=None, auth='Logged In', firstName='Jordyn', gender='F', itemInSession=0, lastName='Jones', length=None, level='free', location='Syracuse, NY', method='GET', page='Home', registration=1513648531284, sessionId=1746, song=None, status=200, ts=1513720913284, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.94 Safari/537.36\"', userId='1747')]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``num`` rows as a :class:`list` of :class:`Row`\n",
    "user_log.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "|summary|           artist|      auth|firstName|gender|     itemInSession|lastName|           length|level|    location|method|   page|        registration|         sessionId|    song|           status|                 ts|           userAgent|            userId|\n",
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "|  count|             8347|     10000|     9664|  9664|             10000|    9664|             8347|10000|        9664| 10000|  10000|                9664|             10000|    8347|            10000|              10000|                9664|             10000|\n",
      "|   mean|            461.0|      null|     null|  null|           19.6734|    null|249.6486587492506| null|        null|  null|   null|1.504695369588739...|         4436.7511|Infinity|         202.8984| 1.5137859954164E12|                null|1442.4413286423842|\n",
      "| stddev|            300.0|      null|     null|  null|25.382114916132608|    null|95.00437130781461| null|        null|  null|   null|  8.47314252131656E9|2043.1281541827561|     NaN|18.04179115450588|3.290828862357974E7|                null| 829.8909432082621|\n",
      "|    min|              !!!|     Guest|   Aakash|     F|                 0| Acevedo|          1.12281| free|Aberdeen, WA|   GET|  About|       1463503881284|                 9|      #1|              200|      1513720872284|\"Mozilla/5.0 (Mac...|                  |\n",
      "|    max|ÃÂlafur Arnalds|Logged Out|     Zoie|     M|               163|  Zuniga|        1806.8371| paid|    Yuma, AZ|   PUT|Upgrade|       1513760702284|              7144|wingless|              404|      1513848349284|Mozilla/5.0 (comp...|               999|\n",
      "+-------+-----------------+----------+---------+------+------------------+--------+-----------------+-----+------------+------+-------+--------------------+------------------+--------+-----------------+-------------------+--------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Computes basic statistics for numeric and string columns, this include count, mean, stddev, min, and max\n",
    "#   If no columns are given, this function computes statistics for all numerical or string columns\n",
    "user_log.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+\n",
      "|summary|           artist|\n",
      "+-------+-----------------+\n",
      "|  count|             8347|\n",
      "|   mean|            461.0|\n",
      "| stddev|            300.0|\n",
      "|    min|              !!!|\n",
      "|    max|ÃÂlafur Arnalds|\n",
      "+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Computes basic statistics for numeric and string columns, this include count, mean, stddev, min, and max\n",
    "#   If no columns are given, this function computes statistics for all numerical or string columns\n",
    "user_log.describe('artist').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|         sessionId|\n",
      "+-------+------------------+\n",
      "|  count|             10000|\n",
      "|   mean|         4436.7511|\n",
      "| stddev|2043.1281541827561|\n",
      "|    min|                 9|\n",
      "|    max|              7144|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Computes basic statistics for numeric and string columns, this include count, mean, stddev, min, and max\n",
    "#   If no columns are given, this function computes statistics for all numerical or string columns\n",
    "user_log.describe('sessionId').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Rows: 10000, Cols: 18\n"
     ]
    }
   ],
   "source": [
    "# Return the number of rows and columns (dimensionality) of the DataFrame\n",
    "print('Rows: {}, Cols: {}'.format(user_log.count(), len(user_log.columns)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|            page|\n",
      "+----------------+\n",
      "|           About|\n",
      "|       Downgrade|\n",
      "|           Error|\n",
      "|            Help|\n",
      "|            Home|\n",
      "|           Login|\n",
      "|          Logout|\n",
      "|        NextSong|\n",
      "|   Save Settings|\n",
      "|        Settings|\n",
      "|Submit Downgrade|\n",
      "|  Submit Upgrade|\n",
      "|         Upgrade|\n",
      "+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "# Return a new DataFrame with duplicate rows removed, optionally only considering certain columns\n",
    "# Returns a new DataFrame sorted by the specified column(s)\n",
    "# Note: cols: list of Column or column names to sort by.\n",
    "#       ascending: boolean or list of boolean (default True); sort ascending vs. descending; specify list for \n",
    "#                  multiple sort orders; if a list is specified, length of the list must equal length of the cols\n",
    "user_log.select('page').dropDuplicates().sort('page').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(userId='1046', firstname='Kenneth', page='NextSong', song='Christmas Tears Will Fall'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Be Wary Of A Woman'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Public Enemy No.1'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Reign Of The Tyrants'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Father And Son'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='No. 5'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Seventeen'),\n",
       " Row(userId='1046', firstname='Kenneth', page='Home', song=None),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='War on war'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Killermont Street'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Black & Blue'),\n",
       " Row(userId='1046', firstname='Kenneth', page='Logout', song=None),\n",
       " Row(userId='1046', firstname='Kenneth', page='Home', song=None),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Heads Will Roll'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Bleed It Out [Live At Milton Keynes]'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Clocks'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Love Rain'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song=\"Ry Ry's Song (Album Version)\"),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='The Invisible Man'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Catch You Baby (Steve Pitron & Max Sanna Radio Edit)'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Ask The Mountains'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Given Up (Album Version)'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='El Cuatrero'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Hero/Heroine'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Spring'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Rising Moon'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Tough Little Boys'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song=\"Qu'Est-Ce Que T'Es Belle\"),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Secrets'),\n",
       " Row(userId='1046', firstname='Kenneth', page='NextSong', song='Under The Gun')]"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "# Filters rows using the given condition\n",
    "user_log.select(['userId', 'firstname', 'page', 'song']).where(user_log.userId == '1046').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculating Statistics by Hour"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a user defined function (UDF)\n",
    "get_hour = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).hour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns a new :class:`DataFrame` by adding a column or replacing the existing column that has the same name\n",
    "user_log = user_log.withColumn('hour', get_hour(user_log.ts))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046', hour='14')"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``n`` rows\n",
    "user_log.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filters rows using the given condition\n",
    "# Groups the DataFrame using the specified columns, so we can run aggregation on them\n",
    "# Returns the number of rows in this DataFrame\n",
    "# Returns a new DataFrame sorted by the specified column(s)\n",
    "# Note: cols: list of Column or column names to sort by.\n",
    "#       ascending: boolean or list of boolean (default True); sort ascending vs. descending; specify list for \n",
    "#                  multiple sort orders; if a list is specified, length of the list must equal length of the cols\n",
    "songs_in_hour = user_log.filter(user_log.page == 'NextSong').groupby(user_log.hour).count().orderBy(user_log.hour.cast('float'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|hour|count|\n",
      "+----+-----+\n",
      "|   0|  375|\n",
      "|   1|  249|\n",
      "|   2|  216|\n",
      "|   3|  228|\n",
      "|   4|  251|\n",
      "|   5|  339|\n",
      "|   6|  462|\n",
      "|   7|  479|\n",
      "|   8|  484|\n",
      "|   9|  430|\n",
      "|  10|  362|\n",
      "|  11|  295|\n",
      "|  12|  257|\n",
      "|  13|  248|\n",
      "|  14|  369|\n",
      "|  15|  375|\n",
      "|  16|  456|\n",
      "|  17|  454|\n",
      "|  18|  382|\n",
      "|  19|  302|\n",
      "+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "songs_in_hour.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns the contents of this :class:`DataFrame` as Pandas ``pandas.DataFrame``\n",
    "songs_in_hour_pd = songs_in_hour.toPandas()\n",
    "\n",
    "# Convert argument to a numeric type\n",
    "songs_in_hour_pd.hour = pd.to_numeric(songs_in_hour_pd.hour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAYUAAAEGCAYAAACKB4k+AAAABHNCSVQICAgIfAhkiAAAAAlwSFlzAAALEgAACxIB0t1+/AAAADh0RVh0U29mdHdhcmUAbWF0cGxvdGxpYiB2ZXJzaW9uMy4xLjEsIGh0dHA6Ly9tYXRwbG90bGliLm9yZy8QZhcZAAAWhUlEQVR4nO3df7DldX3f8ecrC+g2/liBxcLCshiRSVoq4Iq0ZIw/MkGxU7aMVE2iaJhurCQhjaKYMaNNk4h1GlNjpW6qdbH+KFV+TWIUZoGo0/hjEQQdtGwohf0xLAYhUokKvvvH+d7vvSz33j179n7Pz+djZud8v5/zPed89sy553U+38+Pb6oKSZIAfmrUFZAkjQ9DQZLUMhQkSS1DQZLUMhQkSa1DRl2Bg3HkkUfWhg0bRl0NSZooN99883erau1i9010KGzYsIHt27ePuhqSNFGS/N+l7vP0kSSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSpZShIklqGgiSp1WkoJLk7ye1Jbk2yvSk7PMn1Se5sbp/RlCfJ+5PsSHJbktO6rJsk6YmG0VJ4cVWdUlUbm/1LgG1VdSKwrdkHeDlwYvNvM3DZEOomSVpgFKePzgG2NttbgU0Lyi+vni8Da5IcPYL6SdLM6joUCrguyc1JNjdlz6yqPQDN7VFN+Trg3gWP3dmUSZKG5JCOn//Mqtqd5Cjg+iTfXubYLFJWTzioFy6bAdavX78ytZQkAR23FKpqd3O7F7gKOB24b+60UHO7tzl8J3DcgocfC+xe5Dm3VNXGqtq4du3aLqsvSTOns1BI8tNJnjq3DfwS8E3gWuD85rDzgWua7WuB1zWjkM4AHpo7zSRJGo4uTx89E7gqydzrfKKqPpfka8AVSS4A7gHOa47/LHA2sAP4AfCGDuumMXf1Lbt47+e/w+4HH+GYNau5+KyT2HSqXUxS1zoLhaq6C3juIuV/C7x0kfICLuyqPhqNQb7cr75lF2+/8nYe+fFjAOx68BHefuXtAAaD1DFnNKszc1/uux58hGL+y/3qW3Yt+7j3fv47bSDMeeTHj/Hez3+nw9pKAkNBHRr0y333g48cULmklWMoqDODfrkfs2b1AZVLWjmGgjoz6Jf7xWedxOpDVz2ubPWhq7j4rJNWrG6SFmcoqDODfrlvOnUd7z73ZNatWU2AdWtW8+5zT7aTWRqCrmc0a4bNfYkPMrR006nrDAFpBAwF9W2Q4aV+uUuTxVBQX5w7IM0GQ0F9WW54qaEw25x9Pl0MBfXFuQNazKAtSINkfDn6SH1x7oAWM8gExUFnums4DAX1xbkDWswgLUiXMRlvhoL64twBLWaQFqSnIsebfQrqm8NLta+LzzrpcX0KsP8W5DFrVrNrkQDwVOR4sKUgaWCDtCA9FTnebClIOigH2oI8mJnu6p6hIGnoPBU5vjx9JElq2VLQ1HBClHTwDAVNBddmklaGp480FZwQJa0MQ0FTwQlR0sowFDQVXJtJWhmGgqaCE6KklWFHs6aCE6KklWEoaGo4IUo6eJ4+kiS1DAVJUstQkCS1Og+FJKuS3JLkz5v9E5J8JcmdSf5HksOa8ic1+zua+zd0XTdJ0uMNo6VwEXDHgv33AO+rqhOB7wEXNOUXAN+rqmcD72uOkyQNUaehkORY4BXAf232A7wE+HRzyFZgU7N9TrNPc/9Lm+MlSUPSdUvhT4C3Aj9p9o8AHqyqR5v9ncDcGMJ1wL0Azf0PNcc/TpLNSbYn2X7//fd3WXdJmjmdhUKSfw7sraqbFxYvcmj1cd98QdWWqtpYVRvXrl27AjWVJM3pcvLamcC/SHI28GTgafRaDmuSHNK0Bo4FdjfH7wSOA3YmOQR4OvBAh/WTJO2js5ZCVb29qo6tqg3Aq4EbqupXgBuBVzaHnQ9c02xf2+zT3H9DVT2hpSBJ6s4olrl4G/CpJH8A3AJ8uCn/MPCxJDvotRBePYK6SRpjXl2ve0MJhaq6Cbip2b4LOH2RY/4eOG8Y9ZE0eby63nDMzIJ4/sKQljfufyPLXV1vnOo56WYiFPyFIS1vEv5GvLrecMzE2kdev1da3iT8jXh1veFYsqWQ5PDlHlhVEzNc1F8YmjUHeipoEv5GLj7rpMe1ZsCr63VhudNHN9ObPBZgPb11igKsAe4BTui8divkmDWr2bXIh9tfGJpGg5wKmoS/Ea+uNxxLhkJVnQCQ5L8A11bVZ5v9lwO/OJzqrQx/YWgp4965OohBOmQn5W/Eq+t1r5+O5udX1RvndqrqL5P8+w7rtOL8haHFTELn6iAGORXk34jm9BMK303yDuC/0zud9KvA33Zaqw74C0P7mtYhjoOeCvJvZHJ02cLtZ/TRa4C1wFXNv7VNmTTRJqFzdRAXn3USqw9d9biycTwVpMHMtXB3PfgIxXwL9+pbdq3I8++3pdCMMrooyVOq6uEVeVVpDExC5+ogPBV08Ma5r6nrFu5+QyHJP6N3kZynAOuTPBf49ap600G/ujRCk9K5OghPBQ1u3Puaum7h9nP66H3AWTT9CFX1DeCFK/Lq0ghtOnUd7z73ZNatWU2AdWtW8+5zTx6LP3yNzrAn8l19yy7OvPQGTrjkLzjz0hv2exqo60l8fS1zUVX37nNlzMeWOlaaJP6i1r6G2dc0SKuk6xZuPy2Fe5tTSJXksCRvAe5YkVeXpDEzzOU0BmmVdN3C7ael8EbgP9G7hvJO4DrgwhV5dY3EOHeiSaM2zL6mQVslXbZw+wmFnzRXTNMUGPdONGnUhjl6axxHwPUTCl9JcivwEeBzXiJzsk3rhC1pJQ2rr2kcR8D106fwHGAL8DpgR5I/SvKcbqulrkzrhC1pEo3jCLh+Jq8VcD1wfZIX01vu4k1JvgFcUlV/3XEdtYLGsbkqzbJxGwG335ZCkiOSXJRkO/AW4DeBI4E3A5/ouH5aYS6BIGk5/fQp/DXwMWBTVe1cUL69WVZbE8QlECQtJ/vrN06Sce1c3rhxY23fvn3U1ZCkiZLk5qrauNh9/bQUjkzyVuAfAU+eK6yql6xQ/SRJY6Kf0UcfB75N7/Kb/w64G/hah3WSJI1IP6FwRFV9GPhxVf1VVf0acEbH9ZIkjUA/p49+3NzuSfIKYDdwbHdVkiSNSj+h8AdJnk5vCOqfAk8D/m2ntZIkjUQ/k9f+vNl8CHhxt9WRJI3SkqGQ5E+BJYeiVtVvLffESZ4MfAF4UvM6n66qdyY5AfgUcDjwdeC1VfWjJE8CLgeeR++CPq+qqrsP7L8jSY/nqsAHZrmWwsFOAPgh8JKqejjJocCXkvwl8DvA+6rqU83ktwuAy5rb71XVs5O8GngP8KqDrIOkGeaqwAduyVCoqq0L95M8rVdc3+/niZsJbw83u4c2/wp4CfDLTflW4F30QuGcZhvg08AHxnninGabvz4ng6sCH7h+1j7amOR24Dbgm0m+keR5/Tx5klXNstt76S2q9zfAg1X1aHPITnoX76G5vReguf8h4IhFnnNzku1Jtt9///39VENaUXO/Pnc9+AjF/K/P/V1bV8PnqsAHrp95Ch8B3lRVG6rqeHpXXftv/Tx5VT1WVafQG8J6OvCzix3W3GaZ+xY+55aq2lhVG9euXdtPNaQVNewLu2tww7y05rToJxS+X1VfnNupqi8BfZ1CWvCYB4Gb6E16W5Nk7rTVsfTmPUCv1XAcQHP/04EHDuR1pGHw1+fkcFXgA9dPKHw1yYeSvCjJLyT5IHBTktOSnLbUg5KsTbKm2V4N/CJwB3Aj8MrmsPOBa5rta5t9mvtvsD9B48hfn5NjHC9iM+76WSX1xmXurqUWxkvyT+h1JK+iFz5XVNXvJ3kW80NSbwF+tap+2Axh/RhwKr0Wwqur6q7l6uYqqRqFfUe0QO/Xp182mhQHtUpqVQ00Ya2qbqP3Bb9v+V30+hf2Lf974LxBXksaJq9JoWnWzzIXkvYxbpdQ1HiYhqHKhoIkrYBpmSjXT0ezJGk/pmWocj+T185L8tRm+x1Jrlxu1JEkzaJpGarcT0vh96rq+0l+HjiL3oiiy7qtliRNlmkZqtxPKMy1h14BXFZV1wCHdVclSZo80zJRrp+O5l1JPkRv8tl7miWu7YuQpAWmZahyP5PX/gHwMuD2qrozydHAyVV13TAquBwnr0nSgTuoyWvAk+mtW0SSw+ldJ2G5Wc6SFjENY9g1/foJha/TW6jue/RWMl0D7EmyF/jXVXVzh/WTpsK0jGHX9Ounb+BzwNlVdWRVHQG8HLgCeBPwwS4rp/27+pZdnHnpDZxwyV9w5qU3uKb/mJqWMeyafv20FDZW1RvndqrquiR/VFW/03Q6T7VxbvL763NyHMwY9nH+DGr69NNSeCDJ25Ic3/x7K/C9JKuAn3Rcv5Ea9yts+etzcgw6hn3cP4OaPv2Ewi/TuxjO1fSufbC+KVsF/KvuqjZ64/6lOy0zKGfBoGPYx/0zqOnTz9LZ3wV+c4m7d6xsdcbLuH/pHrNmNbsWqcukzaCcBYOOYR/3z6Cmz35DIclzgLcAGxYev9TFdabJuH/pXnzWSYte7GXSZlDOikGW2x73z6CmTz+nj/4nvSukvQO4eMG/qTfu09a91OD0G/fPoKZPP6OPHq2qmVwAbxKmrXuxl+k2CZ9BTZd+lrl4F7AXuIrebGYAquqBTmvWh2lb5sKhh5KG4WCXuTi/uV14yqiAZx1sxTTPOQeSxkE/o49OGEZFZt1yQw8NBUnD0s/oo0OBfwO8sCm6CfhQVf24w3rNHIceShoH/Yw+ugx4Hr11jj7YbM9kx3OXpuWqTZImWz99Cs+vqucu2L8hyTe6qtA0GKTD2DkHksZBP6HwWJKfqaq/AUjyLOYv0al9DNph7NBDSeOgn1C4GLgxyV30rqdwPPCGTms1wQ6mw9g5B5JGrZ/RR9uSnAicRC8Uvl1VP9zPw2aWHcaSJtmSHc1Jnp/kHwI0IXAK8PvAe5vLcmoRdhhLmmTLjT76EPAjgCQvBC4FLgceArZ0X7XJ5Fo1kibZcqGwasFSFq8CtlTVZ6rq94Bn7++JkxyX5MYkdyT5VpKLmvLDk1yf5M7m9hlNeZK8P8mOJLclOe1g/3Oj4CJ1kibZcn0Kq5IcUlWPAi8FNvf5uDmPAm+uqq8neSpwc5LrgdcD26rq0iSXAJcAb6N37ecTm38voDcX4gUH+h8aB3YYS5pUy7UUPgn8VZJrgEeALwIkeTa9U0jLqqo9VfX1Zvv7wB3AOuAcYGtz2FZgU7N9DnB59XwZWJPk6AP/L0mSBrXkL/6q+sMk24CjgetqfjnVn2LpK7EtKskG4FTgK8Azq2pP8xp7khzVHLYOuHfBw3Y2ZXv2ea7NNK2W9evXH0g1JEn7sexpoOYX+75l//tAXiDJU4DPAL9dVX+XZMlDF6vCIq+/haaje+PGjcuv+y1JOiD9rH00sGYxvc8AH6+qK5vi++ZOCzW3e5vyncBxCx5+LLC7y/pJkh6vs1BIr0nwYeCOqvrjBXddy/w1Gs4HrllQ/rpmFNIZwENzp5kkScPRzyiiQZ0JvBa4PcmtTdnv0pvvcEWSC4B7gPOa+z4LnA3sAH6AS2lI0tB1FgpV9SUW7yeA3hDXfY8v4MKu6iNJ2r9O+xQkSZPFUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktQwFSVLLUJAktToLhSQfSbI3yTcXlB2e5Pokdza3z2jKk+T9SXYkuS3JaV3VS5K0tC5bCh8FXrZP2SXAtqo6EdjW7AO8HDix+bcZuKzDekmSltBZKFTVF4AH9ik+B9jabG8FNi0ov7x6vgysSXJ0V3WTJC1u2H0Kz6yqPQDN7VFN+Trg3gXH7WzKniDJ5iTbk2y///77O62sJM2aceloziJltdiBVbWlqjZW1ca1a9d2XC1Jmi3DDoX75k4LNbd7m/KdwHELjjsW2D3kuknSzBt2KFwLnN9snw9cs6D8dc0opDOAh+ZOM0mShueQrp44ySeBFwFHJtkJvBO4FLgiyQXAPcB5zeGfBc4GdgA/AN7QVb0kSUvrLBSq6jVL3PXSRY4t4MKu6iJJ6s+4dDRLksaAoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqSWoSBJahkKkqTWWIVCkpcl+U6SHUkuGXV9JGnWjE0oJFkF/Gfg5cDPAa9J8nOjrZUkzZaxCQXgdGBHVd1VVT8CPgWcM+I6SdJMOWTUFVhgHXDvgv2dwAv2PSjJZmBzs/twku8c4OscCXx3oBpOH9+Leb4X83wv5k3re3H8UneMUyhkkbJ6QkHVFmDLwC+SbK+qjYM+fpr4XszzvZjnezFvFt+LcTp9tBM4bsH+scDuEdVFkmbSOIXC14ATk5yQ5DDg1cC1I66TJM2UsTl9VFWPJvkN4PPAKuAjVfWtDl5q4FNPU8j3Yp7vxTzfi3kz916k6gmn7SVJM2qcTh9JkkbMUJAktWYqFFxGY16Su5PcnuTWJNtHXZ9hSvKRJHuTfHNB2eFJrk9yZ3P7jFHWcViWeC/elWRX89m4NcnZo6zjsCQ5LsmNSe5I8q0kFzXlM/XZmJlQcBmNRb24qk6ZtXHYwEeBl+1TdgmwrapOBLY1+7PgozzxvQB4X/PZOKWqPjvkOo3Ko8Cbq+pngTOAC5vviJn6bMxMKOAyGmpU1ReAB/YpPgfY2mxvBTYNtVIjssR7MZOqak9Vfb3Z/j5wB72VFmbqszFLobDYMhrrRlSXcVDAdUlubpYOmXXPrKo90PtyAI4acX1G7TeS3NacXprq0yWLSbIBOBX4CjP22ZilUOhrGY0ZcmZVnUbvdNqFSV446gppbFwG/AxwCrAH+I+jrc5wJXkK8Bngt6vq70Zdn2GbpVBwGY0Fqmp3c7sXuIre6bVZdl+SowGa270jrs/IVNV9VfVYVf0E+DNm6LOR5FB6gfDxqrqyKZ6pz8YshYLLaDSS/HSSp85tA78EfHP5R029a4Hzm+3zgWtGWJeRmvsCbPxLZuSzkSTAh4E7quqPF9w1U5+NmZrR3Ayt+xPml9H4wxFXaSSSPIte6wB6S518YpbeiySfBF5Eb1nk+4B3AlcDVwDrgXuA86pq6jtgl3gvXkTv1FEBdwO/PndOfZol+Xngi8DtwE+a4t+l168wM5+NmQoFSdLyZun0kSRpPwwFSVLLUJAktQwFSVLLUJAktQwFqU9JHt5n//VJPjCq+khdMBSkEWtW8JXGgqEgrYAkxyfZ1iwity3J+qb8o0leueC4h5vbFzVr93+C3mQpaSwcMuoKSBNkdZJbF+wfzvxSKR8ALq+qrUl+DXg/+19i+XTgH1fV/1n5qkqDMRSk/j1SVafM7SR5PTB3gaJ/CpzbbH8M+A99PN9XDQSNG08fSd2YWz/mUZq/s2bBtcMWHPP/hl0paX8MBWll/C96K+8C/ArwpWb7buB5zfY5wKHDrZZ0YAwFaWX8FvCGJLcBrwUuasr/DPiFJF8FXoCtA405V0mVJLVsKUiSWoaCJKllKEiSWoaCJKllKEiSWoaCJKllKEiSWv8fuf7efL0stH8AAAAASUVORK5CYII=\n",
      "text/plain": [
       "<Figure size 432x288 with 1 Axes>"
      ]
     },
     "metadata": {
      "needs_background": "light"
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create a scatter plot\n",
    "plt.scatter(songs_in_hour_pd['hour'], songs_in_hour_pd['count'])\n",
    "plt.xlim(-1, 24);\n",
    "plt.ylim(0, 1.2 * max(songs_in_hour_pd['count']))\n",
    "plt.xlabel('Hour')\n",
    "plt.ylabel('Songs played');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop Rows with Missing Values\n",
    "\n",
    "As you'll see, it turns out there are no missing values in the userID or session columns. But there are userID values that are empty strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns a new :class:`DataFrame` omitting rows with null values\n",
    "# Note: how: if 'any', drop a row if it contains any nulls; if 'all', drop a row only if all its values are null\n",
    "#       subset: optional list of column names to consider\n",
    "user_log_valid = user_log.dropna(how='any', subset=['userId', 'sessionId'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the number of rows in this DataFrame\n",
    "user_log_valid.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|      |\n",
      "|    10|\n",
      "|   100|\n",
      "|  1000|\n",
      "|  1003|\n",
      "|  1005|\n",
      "|  1006|\n",
      "|  1017|\n",
      "|  1019|\n",
      "|  1020|\n",
      "|  1022|\n",
      "|  1025|\n",
      "|  1030|\n",
      "|  1035|\n",
      "|  1037|\n",
      "|   104|\n",
      "|  1040|\n",
      "|  1042|\n",
      "|  1043|\n",
      "|  1046|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "# Return a new DataFrame with duplicate rows removed, optionally only considering certain columns\n",
    "# Returns a new DataFrame sorted by the specified column(s)\n",
    "# Note: cols: list of Column or column names to sort by.\n",
    "#       ascending: boolean or list of boolean (default True); sort ascending vs. descending; specify list for \n",
    "#                  multiple sort orders; if a list is specified, length of the list must equal length of the cols\n",
    "user_log.select('userId').dropDuplicates().sort('userId').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9664"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Filters rows using the given condition\n",
    "user_log_valid = user_log_valid.filter(user_log_valid['userId'] != '')\n",
    "\n",
    "# Returns the number of rows in this DataFrame\n",
    "user_log_valid.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9664"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Filters rows using the given condition\n",
    "user_log_valid = user_log_valid.where(user_log_valid.userId != '')\n",
    "\n",
    "# Returns the number of rows in this DataFrame\n",
    "user_log_valid.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Users Downgrade Their Accounts\n",
    "\n",
    "Find when users downgrade their accounts and then flag those log entries. Then use a window function and cumulative sum to distinguish each user's data as either pre or post downgrade events."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+---------+------+-------------+--------+------+-----+--------------------+------+----------------+-------------+---------+----+------+-------------+--------------------+------+----+\n",
      "|artist|     auth|firstName|gender|itemInSession|lastName|length|level|            location|method|            page| registration|sessionId|song|status|           ts|           userAgent|userId|hour|\n",
      "+------+---------+---------+------+-------------+--------+------+-----+--------------------+------+----------------+-------------+---------+----+------+-------------+--------------------+------+----+\n",
      "|  null|Logged In|    Kelly|     F|           24|  Newton|  null| paid|Houston-The Woodl...|   PUT|Submit Downgrade|1513283366284|     5931|null|   307|1513768454284|Mozilla/5.0 (Wind...|  1138|   3|\n",
      "+------+---------+---------+------+-------------+--------+------+-----+--------------------+------+----------------+-------------+---------+----+------+-------------+--------------------+------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filters rows using the given condition\n",
    "user_log_valid.filter(user_log_valid.page == 'Submit Downgrade').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+--------+-----+---------------------------------------+\n",
      "|userId|firstname|page    |level|song                                   |\n",
      "+------+---------+--------+-----+---------------------------------------+\n",
      "|1138  |Kelly    |Home    |paid |null                                   |\n",
      "|1138  |Kelly    |NextSong|paid |Everybody Everybody                    |\n",
      "|1138  |Kelly    |NextSong|paid |Gears                                  |\n",
      "|1138  |Kelly    |NextSong|paid |Use Somebody                           |\n",
      "|1138  |Kelly    |NextSong|paid |Love Of My Life (1993 Digital Remaster)|\n",
      "|1138  |Kelly    |NextSong|paid |Down In The Valley Woe                 |\n",
      "|1138  |Kelly    |NextSong|paid |Treat Her Like A Lady                  |\n",
      "|1138  |Kelly    |NextSong|paid |Everybody Thinks You're An Angel       |\n",
      "|1138  |Kelly    |NextSong|paid |Fourteen Wives                         |\n",
      "|1138  |Kelly    |NextSong|paid |Love On The Rocks                      |\n",
      "|1138  |Kelly    |NextSong|paid |Breakeven                              |\n",
      "|1138  |Kelly    |NextSong|paid |Leaf House                             |\n",
      "|1138  |Kelly    |NextSong|paid |NAISEN KANSSA                          |\n",
      "|1138  |Kelly    |NextSong|paid |You're In My Heart                     |\n",
      "|1138  |Kelly    |NextSong|paid |Roll On Down The Highway               |\n",
      "|1138  |Kelly    |NextSong|paid |Plasticities (Remix)                   |\n",
      "|1138  |Kelly    |NextSong|paid |Secrets                                |\n",
      "|1138  |Kelly    |NextSong|paid |Hello                                  |\n",
      "|1138  |Kelly    |NextSong|paid |I Never Told You                       |\n",
      "|1138  |Kelly    |NextSong|paid |Love Break Me                          |\n",
      "+------+---------+--------+-----+---------------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "# Filters rows using the given condition\n",
    "user_log.select(['userId', 'firstname', 'page', 'level', 'song']).where(user_log.userId == '1138').show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(userId='1138', firstname='Kelly', page='Home', level='paid', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Everybody Everybody'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Gears'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Use Somebody'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Love Of My Life (1993 Digital Remaster)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Down In The Valley Woe'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Treat Her Like A Lady'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"Everybody Thinks You're An Angel\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Fourteen Wives'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Love On The Rocks'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Breakeven'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Leaf House'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='NAISEN KANSSA'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"You're In My Heart\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Roll On Down The Highway'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Plasticities (Remix)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Secrets'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Hello'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='I Never Told You'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Love Break Me'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='One Touch One Bounce'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Undo'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Overdue (Blackbeard Remix)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Slave To Love (1999 Digital Remaster)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Stronger'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='All Of Us (Album Version)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Sehr kosmisch'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='March Of The Celts'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Electricity'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Aces High'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Bananeira'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='The General'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='HÃ\\x83Â©roe De Leyenda (VersiÃ\\x83Â³n Maxi)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"Don't Stop The Music\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"You're The One\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Entering White Cecilia'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Piccolo Cesare'),\n",
       " Row(userId='1138', firstname='Kelly', page='Help', level='paid', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Last Christmas (Album Version)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='You Shook Me'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Going Steady'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='My Name Is'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Undo'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Secrets'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Good Times Gone (Album Version)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Angelito'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Batdance ( LP Version )'),\n",
       " Row(userId='1138', firstname='Kelly', page='Home', level='paid', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='DiÃ\\x83Â¡kdal'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Whirring'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Potholderz (feat. Count Bass D)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Seaside'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Louder Than A Bomb'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Just Like You'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"You're The One\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Turn It Again (Album Version)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Everywhere I Go'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song=\"Easy Skankin'\"),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Roses'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Killing Me Softly With His Song'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='The Razor (Album Version)'),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='paid', song='Idols and Anchors'),\n",
       " Row(userId='1138', firstname='Kelly', page='Downgrade', level='paid', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='Submit Downgrade', level='paid', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='Home', level='free', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='free', song='Bones'),\n",
       " Row(userId='1138', firstname='Kelly', page='Home', level='free', song=None),\n",
       " Row(userId='1138', firstname='Kelly', page='NextSong', level='free', song='Grenouilles Mantidactylus (Small Frogs)')]"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log.select(['userId', 'firstname', 'page', 'level', 'song']).where(user_log.userId == '1138').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a user defined function (UDF)\n",
    "flag_downgrade_event = udf(lambda x: 1 if x == 'Submit Downgrade' else 0, IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns a new :class:`DataFrame` by adding a column or replacing the existing column that has the same name\n",
    "user_log_valid = user_log_valid.withColumn('downgraded', flag_downgrade_event('page'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046', hour='14', downgraded=0)"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``n`` rows\n",
    "user_log_valid.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creates a :class:`WindowSpec` with the partitioning defined\n",
    "windowval = Window.partitionBy('userId').orderBy(desc('ts')).rangeBetween(Window.unboundedPreceding, 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns a new :class:`DataFrame` by adding a column or replacing the existing column that has the same name\n",
    "user_log_valid = user_log_valid.withColumn('phase', Fsum('downgraded').over(windowval))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(userId='1138', firstname='Kelly', ts=1513729066284, page='Home', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513729066284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513729313284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513729552284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513729783284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513730001284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513730263284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513730518284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513730768284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513731182284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513731435284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513731695284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513731857284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513732160284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513732302284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513732540284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513732770284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513732994284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513733223284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513733456284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513733738284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513733941284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513734289284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513734598284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513734863284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513735174284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513735385284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513736040284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513736237284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513736446284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513736709284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513736915284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513737160284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513737460284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513737728284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513737936284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513738144284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513738197284, page='Help', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513738432284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513763195284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513763453284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513763622284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513763890284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513764238284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513764462284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513764782284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513765075284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513765075284, page='Home', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513765448284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513765655284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513765818284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513766091284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513766189284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513766385284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513766599284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513766838284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513767203284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513767413284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513767643284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768012284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768242284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768452284, page='NextSong', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768453284, page='Downgrade', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768454284, page='Submit Downgrade', level='paid', phase=1),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513768456284, page='Home', level='free', phase=0),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513814880284, page='NextSong', level='free', phase=0),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513821430284, page='Home', level='free', phase=0),\n",
       " Row(userId='1138', firstname='Kelly', ts=1513833144284, page='NextSong', level='free', phase=0)]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Projects a set of expressions and returns a new :class:`DataFrame\n",
    "# Note: cols: list of column names (string) or expressions (:class:`Column`); if one of the column names is '*', \n",
    "#             that column is expanded to include all columns in the current :class:`DataFrame`\n",
    "# Filters rows using the given condition\n",
    "# Returns a new DataFrame sorted by the specified column(s)\n",
    "# Note: cols: list of Column or column names to sort by.\n",
    "#       ascending: boolean or list of boolean (default True); sort ascending vs. descending; specify list for \n",
    "#                  multiple sort orders; if a list is specified, length of the list must equal length of the cols\n",
    "user_log_valid.select(['userId', 'firstname', 'ts', 'page', 'level', 'phase']).where(user_log.userId == '1138').sort('ts').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 1\n",
    "\n",
    "Which page did user id '' (empty string) NOT visit?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(userId='', page='About'),\n",
       " Row(userId='', page='Help'),\n",
       " Row(userId='', page='Home'),\n",
       " Row(userId='', page='Login')]"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log.select(['userId', 'page',]).dropDuplicates().sort('page').where(user_log.userId == '').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(blank_pages='About'),\n",
       " Row(blank_pages='Help'),\n",
       " Row(blank_pages='Home'),\n",
       " Row(blank_pages='Login')]"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Filter for users with blank user id\n",
    "blank_pages = user_log.where(user_log.userId == '') \\\n",
    "    .select(col('page') \\\n",
    "    .alias('blank_pages')) \\\n",
    "    .dropDuplicates() \\\n",
    "    .sort('blank_pages')\n",
    "blank_pages.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|            page|\n",
      "+----------------+\n",
      "|           About|\n",
      "|       Downgrade|\n",
      "|           Error|\n",
      "|            Help|\n",
      "|            Home|\n",
      "|           Login|\n",
      "|          Logout|\n",
      "|        NextSong|\n",
      "|   Save Settings|\n",
      "|        Settings|\n",
      "|Submit Downgrade|\n",
      "|  Submit Upgrade|\n",
      "|         Upgrade|\n",
      "+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Get a list of possible pages that could be visited\n",
    "all_pages = user_log.select('page').dropDuplicates().sort('page')\n",
    "all_pages.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NextSong\n",
      "Upgrade\n",
      "Save Settings\n",
      "Error\n",
      "Logout\n",
      "Downgrade\n",
      "Settings\n",
      "Submit Upgrade\n",
      "Submit Downgrade\n"
     ]
    }
   ],
   "source": [
    "# Find values in all_pages that are not in blank_pages\n",
    "# These are the pages that the blank user did not go to\n",
    "for row in set(all_pages.collect()) - set(blank_pages.collect()):\n",
    "    print(row.page)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 2\n",
    "\n",
    "What type of user does the empty string userId most likely refer to?\n",
    "\n",
    "Perhaps it represents users who have not signed up yet or who are signed out and are about to log in."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 3\n",
    "\n",
    "How many female users do we have in the data set?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3820"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# All female user records\n",
    "user_log.select('userId', 'gender') \\\n",
    "    .where(user_log.gender=='F') \\\n",
    "    .count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "462"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Unique female user records\n",
    "user_log.select('userId', 'gender') \\\n",
    "    .dropDuplicates() \\\n",
    "    .where(user_log.gender=='F') \\\n",
    "    .count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 4\n",
    "\n",
    "How many songs were played from the most played artist?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|  artist|count|\n",
      "+--------+-----+\n",
      "|Coldplay|   83|\n",
      "+--------+-----+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_log.where(user_log.page == \"NextSong\").groupby(user_log.artist).count().sort('count', ascending=False).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------------+\n",
      "|  artist|artist_count|\n",
      "+--------+------------+\n",
      "|Coldplay|          83|\n",
      "+--------+------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_log.select('artist') \\\n",
    "    .where(user_log.page == 'NextSong') \\\n",
    "    .groupby('artist') \\\n",
    "    .agg({'artist':'count'}) \\\n",
    "    .withColumnRenamed('count(artist)', 'artist_count') \\\n",
    "    .sort(desc('artist_count')) \\\n",
    "    .show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 5\n",
    "\n",
    "How many songs do users listen to on average between visiting our home page? Please round your answer to the closest integer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|avg(count(period))|\n",
      "+------------------+\n",
      "| 6.898347107438017|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "function = udf(lambda ishome : int(ishome == 'Home'), IntegerType())\n",
    "\n",
    "user_window = Window \\\n",
    "    .partitionBy('userID') \\\n",
    "    .orderBy(desc('ts')) \\\n",
    "    .rangeBetween(Window.unboundedPreceding, 0)\n",
    "\n",
    "cusum = user_log.where((user_log.page == 'NextSong') | (user_log.page == 'Home')) \\\n",
    "    .select('userID', 'page', 'ts') \\\n",
    "    .withColumn('homevisit', function(col('page'))) \\\n",
    "    .withColumn('period', Fsum('homevisit').over(user_window))\n",
    "\n",
    "cusum.where((cusum.page == 'NextSong')) \\\n",
    "    .groupBy('userID', 'period') \\\n",
    "    .agg({'period':'count'}) \\\n",
    "    .agg({'count(period)':'avg'}).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Wrangling with Spark SQL\n",
    "\n",
    "Helpful resources:\n",
    "http://spark.apache.org/docs/latest/api/python/pyspark.sql.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql.functions import asc, desc\n",
    "from pyspark.sql.functions import sum as Fsum\n",
    "\n",
    "import datetime\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Instantiate a Spark session \n",
    "# The entry point to programming Spark with the Dataset and DataFrame API\n",
    "# Note: appName(): sets a name for the application, which will be shown in the Spark web UI\n",
    "#       getOrCreate(): get or instantiate a SparkContext and register it as a singleton object\n",
    "spark = SparkSession.builder \\\n",
    "    .appName('Data wrangling with Spark SQL') \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = '/Users/yangweichle/Documents/Employment/TRAINING/DATA SCIENCE/Spark/Udacity_Spark for Big Data/Data Wrangling with Spark/data/sparkify_log_small.json'\n",
    "\n",
    "# Loads JSON files and returns the results as a :class:`DataFrame`\n",
    "# Note: path: string represents path to the JSON dataset, or a list of paths, or RDD of Strings storing JSON objects\n",
    "user_log = spark.read.json(path=path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046')]"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Returns the first ``num`` rows as a :class:`list` of :class:`Row`\n",
    "user_log.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: long (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prints out the schema in the tree format\n",
    "user_log.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a View And Run Queries\n",
    "\n",
    "The code below creates a temporary view against which you can run SQL queries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creates or replaces a local temporary view with this :class:`DataFrame`\n",
    "user_log.createOrReplaceTempView('user_log_view')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|       artist|     auth|firstName|gender|itemInSession|lastName|   length|level|            location|method|    page| registration|sessionId|                song|status|           ts|           userAgent|userId|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|Showaddywaddy|Logged In|  Kenneth|     M|          112|Matthews|232.93342| paid|Charlotte-Concord...|   PUT|NextSong|1509380319284|     5132|Christmas Tears W...|   200|1513720872284|\"Mozilla/5.0 (Win...|  1046|\n",
      "|   Lily Allen|Logged In|Elizabeth|     F|            7|   Chase|195.23873| free|Shreveport-Bossie...|   PUT|NextSong|1512718541284|     5027|       Cheryl Tweedy|   200|1513720878284|\"Mozilla/5.0 (Win...|  1000|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('SELECT * FROM user_log_view LIMIT 2').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|       artist|     auth|firstName|gender|itemInSession|lastName|   length|level|            location|method|    page| registration|sessionId|                song|status|           ts|           userAgent|userId|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "|Showaddywaddy|Logged In|  Kenneth|     M|          112|Matthews|232.93342| paid|Charlotte-Concord...|   PUT|NextSong|1509380319284|     5132|Christmas Tears W...|   200|1513720872284|\"Mozilla/5.0 (Win...|  1046|\n",
      "|   Lily Allen|Logged In|Elizabeth|     F|            7|   Chase|195.23873| free|Shreveport-Bossie...|   PUT|NextSong|1512718541284|     5027|       Cheryl Tweedy|   200|1513720878284|\"Mozilla/5.0 (Win...|  1000|\n",
      "+-------------+---------+---------+------+-------------+--------+---------+-----+--------------------+------+--------+-------------+---------+--------------------+------+-------------+--------------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "          SELECT * \n",
    "          FROM user_log_view \n",
    "          LIMIT 2\n",
    "          '''\n",
    "          ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|   10000|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "          SELECT COUNT(*) \n",
    "          FROM user_log_view \n",
    "          '''\n",
    "          ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+--------+----------------------------------------------------+\n",
      "|userID|firstname|page    |song                                                |\n",
      "+------+---------+--------+----------------------------------------------------+\n",
      "|1046  |Kenneth  |NextSong|Christmas Tears Will Fall                           |\n",
      "|1046  |Kenneth  |NextSong|Be Wary Of A Woman                                  |\n",
      "|1046  |Kenneth  |NextSong|Public Enemy No.1                                   |\n",
      "|1046  |Kenneth  |NextSong|Reign Of The Tyrants                                |\n",
      "|1046  |Kenneth  |NextSong|Father And Son                                      |\n",
      "|1046  |Kenneth  |NextSong|No. 5                                               |\n",
      "|1046  |Kenneth  |NextSong|Seventeen                                           |\n",
      "|1046  |Kenneth  |Home    |null                                                |\n",
      "|1046  |Kenneth  |NextSong|War on war                                          |\n",
      "|1046  |Kenneth  |NextSong|Killermont Street                                   |\n",
      "|1046  |Kenneth  |NextSong|Black & Blue                                        |\n",
      "|1046  |Kenneth  |Logout  |null                                                |\n",
      "|1046  |Kenneth  |Home    |null                                                |\n",
      "|1046  |Kenneth  |NextSong|Heads Will Roll                                     |\n",
      "|1046  |Kenneth  |NextSong|Bleed It Out [Live At Milton Keynes]                |\n",
      "|1046  |Kenneth  |NextSong|Clocks                                              |\n",
      "|1046  |Kenneth  |NextSong|Love Rain                                           |\n",
      "|1046  |Kenneth  |NextSong|Ry Ry's Song (Album Version)                        |\n",
      "|1046  |Kenneth  |NextSong|The Invisible Man                                   |\n",
      "|1046  |Kenneth  |NextSong|Catch You Baby (Steve Pitron & Max Sanna Radio Edit)|\n",
      "|1046  |Kenneth  |NextSong|Ask The Mountains                                   |\n",
      "|1046  |Kenneth  |NextSong|Given Up (Album Version)                            |\n",
      "|1046  |Kenneth  |NextSong|El Cuatrero                                         |\n",
      "|1046  |Kenneth  |NextSong|Hero/Heroine                                        |\n",
      "|1046  |Kenneth  |NextSong|Spring                                              |\n",
      "|1046  |Kenneth  |NextSong|Rising Moon                                         |\n",
      "|1046  |Kenneth  |NextSong|Tough Little Boys                                   |\n",
      "|1046  |Kenneth  |NextSong|Qu'Est-Ce Que T'Es Belle                            |\n",
      "|1046  |Kenneth  |NextSong|Secrets                                             |\n",
      "|1046  |Kenneth  |NextSong|Under The Gun                                       |\n",
      "+------+---------+--------+----------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "          SELECT userID,\n",
    "                 firstname,\n",
    "                 page,\n",
    "                 song\n",
    "          FROM user_log_view \n",
    "          WHERE userID == '1046'\n",
    "          '''\n",
    "          ).show(100, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|            page|\n",
      "+----------------+\n",
      "|           About|\n",
      "|       Downgrade|\n",
      "|           Error|\n",
      "|            Help|\n",
      "|            Home|\n",
      "|           Login|\n",
      "|          Logout|\n",
      "|        NextSong|\n",
      "|   Save Settings|\n",
      "|        Settings|\n",
      "|Submit Downgrade|\n",
      "|  Submit Upgrade|\n",
      "|         Upgrade|\n",
      "+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "          SELECT DISTINCT page\n",
    "          FROM user_log_view \n",
    "          ORDER BY page ASC\n",
    "          '''\n",
    "          ).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### User Defined Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function __main__.<lambda>(x)>"
      ]
     },
     "execution_count": 82,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Register a Python function (including lambda function) or a user-defined function as a SQL function\n",
    "spark.udf.register('get_hour', lambda x: int(datetime.datetime.fromtimestamp(x / 1000.0).hour))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(artist='Showaddywaddy', auth='Logged In', firstName='Kenneth', gender='M', itemInSession=112, lastName='Matthews', length=232.93342, level='paid', location='Charlotte-Concord-Gastonia, NC-SC', method='PUT', page='NextSong', registration=1509380319284, sessionId=5132, song='Christmas Tears Will Fall', status=200, ts=1513720872284, userAgent='\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='1046', hour='14')]"
      ]
     },
     "execution_count": 83,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql('''\n",
    "          SELECT *,\n",
    "                 get_hour(ts) AS hour\n",
    "          FROM user_log_view\n",
    "          LIMIT 1\n",
    "          '''\n",
    "          ).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [],
   "source": [
    "songs_in_hour = spark.sql('''\n",
    "          SELECT get_hour(ts) AS hour,\n",
    "                 COUNT(*) AS plays_per_hour\n",
    "          FROM user_log_view\n",
    "          WHERE page = \"NextSong\"\n",
    "          GROUP BY hour\n",
    "          ORDER BY CAST(hour AS int) ASC\n",
    "          '''\n",
    "          )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+\n",
      "|hour|plays_per_hour|\n",
      "+----+--------------+\n",
      "|   0|           375|\n",
      "|   1|           249|\n",
      "|   2|           216|\n",
      "|   3|           228|\n",
      "|   4|           251|\n",
      "|   5|           339|\n",
      "|   6|           462|\n",
      "|   7|           479|\n",
      "|   8|           484|\n",
      "|   9|           430|\n",
      "|  10|           362|\n",
      "|  11|           295|\n",
      "|  12|           257|\n",
      "|  13|           248|\n",
      "|  14|           369|\n",
      "|  15|           375|\n",
      "|  16|           456|\n",
      "|  17|           454|\n",
      "|  18|           382|\n",
      "|  19|           302|\n",
      "|  20|           352|\n",
      "|  21|           276|\n",
      "|  22|           348|\n",
      "|  23|           358|\n",
      "+----+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "songs_in_hour.show(n=25)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Converting Results to Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Returns the contents of this :class:`DataFrame` as Pandas ``pandas.DataFrame``\n",
    "songs_in_hour_pd = songs_in_hour.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   hour  plays_per_hour\n",
      "0     0             375\n",
      "1     1             249\n",
      "2     2             216\n",
      "3     3             228\n",
      "4     4             251\n",
      "5     5             339\n",
      "6     6             462\n",
      "7     7             479\n",
      "8     8             484\n",
      "9     9             430\n",
      "10   10             362\n",
      "11   11             295\n",
      "12   12             257\n",
      "13   13             248\n",
      "14   14             369\n",
      "15   15             375\n",
      "16   16             456\n",
      "17   17             454\n",
      "18   18             382\n",
      "19   19             302\n",
      "20   20             352\n",
      "21   21             276\n",
      "22   22             348\n",
      "23   23             358\n"
     ]
    }
   ],
   "source": [
    "print(songs_in_hour_pd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 1\n",
    "\n",
    "Which page did user id '' (empty string) NOT visit?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------------+\n",
      "|page|            page|\n",
      "+----+----------------+\n",
      "|null|Submit Downgrade|\n",
      "|null|       Downgrade|\n",
      "|null|          Logout|\n",
      "|null|   Save Settings|\n",
      "|null|        Settings|\n",
      "|null|        NextSong|\n",
      "|null|         Upgrade|\n",
      "|null|           Error|\n",
      "|null|  Submit Upgrade|\n",
      "+----+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SELECT distinct pages for the blank user and distinct pages for all users\n",
    "# Right join the results to find pages that blank visitor did not visit\n",
    "spark.sql(\"\"\"\n",
    "          SELECT *\n",
    "          FROM (SELECT DISTINCT page\n",
    "                FROM user_log_view\n",
    "                WHERE userID = '') AS user_pages\n",
    "          RIGHT JOIN (SELECT DISTINCT page\n",
    "                      FROM user_log_view) AS all_pages\n",
    "          ON user_pages.page = all_pages.page\n",
    "          WHERE user_pages.page IS NULL\n",
    "          \"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 2\n",
    "\n",
    "Why might you prefer to use SQL over DataFrames? Why might you prefer DataFrames over SQL?\n",
    "\n",
    "Both Spark SQL and Spark DataFrames are part of the Spark SQL library. Hence, they both use the Spark SQL Catalyst Optimizer to optimize queries. \n",
    "\n",
    "You might prefer SQL over DataFrames because the syntax is clearer especially for teams already experienced in SQL.\n",
    "\n",
    "Spark DataFrames give you more control. You can break down your queries into smaller steps, which can make debugging easier. You can also [cache](https://unraveldata.com/to-cache-or-not-to-cache/) intermediate results or [repartition](https://hackernoon.com/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4) intermediate results."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 3\n",
    "\n",
    "How many female users do we have in the data set?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+\n",
      "|count(DISTINCT userID)|\n",
      "+----------------------+\n",
      "|                   462|\n",
      "+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "          SELECT COUNT(DISTINCT userID)\n",
    "          FROM user_log_view\n",
    "          WHERE gender = 'F'\n",
    "          \"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 4\n",
    "\n",
    "How many songs were played from the most played artist?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|  artist|plays|\n",
      "+--------+-----+\n",
      "|Coldplay|   83|\n",
      "+--------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Here is one solution\n",
    "spark.sql(\"\"\"\n",
    "          SELECT artist,\n",
    "                 COUNT(artist) AS plays\n",
    "          FROM user_log_view\n",
    "          GROUP BY Artist\n",
    "          ORDER BY plays DESC\n",
    "          LIMIT 1\n",
    "          \"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|  artist|plays|\n",
      "+--------+-----+\n",
      "|Coldplay|   83|\n",
      "+--------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Here is an alternative solution\n",
    "# Get the artist play counts\n",
    "play_counts = \\\n",
    "spark.sql(\"\"\"\n",
    "          SELECT artist,\n",
    "                 COUNT(artist) AS plays\n",
    "          FROM user_log_view\n",
    "          GROUP BY artist\n",
    "          \"\"\")\n",
    "\n",
    "# save the results in a new view\n",
    "play_counts.createOrReplaceTempView('artist_counts')\n",
    "\n",
    "# use a self join to find where the max play equals the count value\n",
    "spark.sql(\"\"\"\n",
    "          SELECT a2.artist, a2.plays\n",
    "          FROM (SELECT max(plays) AS max_plays\n",
    "                FROM artist_counts) AS a1\n",
    "          INNER JOIN artist_counts AS a2\n",
    "          ON a1.max_plays = a2.plays\n",
    "          \"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 5\n",
    "\n",
    "How many songs do users listen to on average between visiting our home page? Please round your answer to the closest integer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|avg(count_results)|\n",
      "+------------------+\n",
      "| 6.898347107438017|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SELECT CASE WHEN 1 > 0 THEN 1 WHEN 2 > 0 THEN 2.0 ELSE 1.2 END;\n",
    "is_home = \\\n",
    "spark.sql(\"\"\"\n",
    "          SELECT userID,\n",
    "                 page,\n",
    "                 ts,\n",
    "                 CASE WHEN page = 'Home' THEN 1\n",
    "                      ELSE 0 END AS is_home\n",
    "          FROM user_log_view\n",
    "          WHERE (page = 'NextSong') OR (page = 'Home')\n",
    "          \"\"\")\n",
    "\n",
    "# keep the results in a new view\n",
    "is_home.createOrReplaceTempView('is_home_view')\n",
    "\n",
    "# find the cumulative sum over the is_home column\n",
    "cumulative_sum = \\\n",
    "spark.sql(\"\"\"\n",
    "          SELECT *,\n",
    "                 SUM(is_home) OVER (PARTITION BY userID ORDER BY ts DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS period\n",
    "          FROM is_home_view\n",
    "          \"\"\")\n",
    "\n",
    "# keep the results in a view\n",
    "cumulative_sum.createOrReplaceTempView('period_view')\n",
    "\n",
    "# find the average count for NextSong\n",
    "spark.sql(\"\"\"\n",
    "          SELECT AVG(count_results)\n",
    "          FROM (SELECT COUNT(*) AS count_results\n",
    "                FROM period_view\n",
    "                GROUP BY userID,\n",
    "                         period,\n",
    "                         page\n",
    "                HAVING page = 'NextSong') AS counts\n",
    "          \"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}