4_data_wrangling.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Wrangling with Spark\n",
    "\n",
    "This is the code used in the previous screencast. Run each code cell to understand what the code does and how it works.\n",
    "\n",
    "These first three cells import libraries, instantiate a SparkSession, and then read in the data set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "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 desc\n",
    "from pyspark.sql.functions import asc\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": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Wrangling Data\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = \"data/sparkify_log_small.json\"\n",
    "user_log = spark.read.json(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Exploration \n",
    "\n",
    "The next cells explore the data set."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "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": [
    "user_log.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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.6486587492503| null|        null|  null|   null|1.504695369588739...|         4436.7511|Infinity|          202.8984|  1.5137859954164E12|                null|1442.4413286423842|\n",
      "| stddev|            300.0|      null|     null|  null|25.382114916132597|    null|95.00437130781461| null|        null|  null|   null|  8.47314252131657E9|2043.1281541827557|     NaN|18.041791154505876|3.2908288623601213E7|                null| 829.8909432082613|\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": [
    "user_log.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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": [
    "user_log.describe(\"artist\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|         sessionId|\n",
      "+-------+------------------+\n",
      "|  count|             10000|\n",
      "|   mean|         4436.7511|\n",
      "| stddev|2043.1281541827557|\n",
      "|    min|                 9|\n",
      "|    max|              7144|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_log.describe(\"sessionId\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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": [
    "user_log.select(\"page\").dropDuplicates().sort(\"page\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "get_hour = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0). hour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_log = user_log.withColumn(\"hour\", get_hour(user_log.ts))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "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='22')"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "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": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|hour|count|\n",
      "+----+-----+\n",
      "|   0|  456|\n",
      "|   1|  454|\n",
      "|   2|  382|\n",
      "|   3|  302|\n",
      "|   4|  352|\n",
      "|   5|  276|\n",
      "|   6|  348|\n",
      "|   7|  358|\n",
      "|   8|  375|\n",
      "|   9|  249|\n",
      "|  10|  216|\n",
      "|  11|  228|\n",
      "|  12|  251|\n",
      "|  13|  339|\n",
      "|  14|  462|\n",
      "|  15|  479|\n",
      "|  16|  484|\n",
      "|  17|  430|\n",
      "|  18|  362|\n",
      "|  19|  295|\n",
      "+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "songs_in_hour.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "songs_in_hour_pd = songs_in_hour.toPandas()\n",
    "songs_in_hour_pd.hour = pd.to_numeric(songs_in_hour_pd.hour)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAYgAAAEKCAYAAAAIO8L1AAAABHNCSVQICAgIfAhkiAAAAAlwSFlzAAALEgAACxIB0t1+/AAAADl0RVh0U29mdHdhcmUAbWF0cGxvdGxpYiB2ZXJzaW9uIDIuMS4wLCBodHRwOi8vbWF0cGxvdGxpYi5vcmcvpW3flQAAFo1JREFUeJzt3X+w5XV93/HnKwvoNhpXYKGwsCzGldGWCrhSWjIGNRMEO4UyUjWJbgzTjZUkpCqKGTPatIlYp8EaK3ETqIv1R6nyaxKiMAtEncYfiyDgIGVDKeyPYSH8CFSi/Hj3j/O9udfle+9+9+w9P+45z8fMnfP9fs73nPu5Z849r/P5fD+fzzdVhSRJu/upUVdAkjSeDAhJUisDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1MiAkSa32G3UF9sXBBx9ca9asGXU1JGlJufnmmx+qqpV7Om5JB8SaNWvYsmXLqKshSUtKkv/b5Ti7mCRJrQwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQYaEEnuTXJ7kluTbGnKDkxyfZK7m9sXN+VJ8okkW5PcluSEQdZNkrSwYbQgXltVx1XVumb/AmBzVa0FNjf7AKcBa5ufDcDFQ6ibJGkeo+hiOgPY1GxvAs6cU35Z9XwTWJHksBHUT5LE4AOigOuS3JxkQ1N2aFXtBGhuD2nKVwH3z3nstqZMkjQC+w34+U+uqh1JDgGuT/KDBY5NS1k956Be0GwAWL169eLUUpL0HANtQVTVjuZ2F3AlcCLwwEzXUXO7qzl8G3DknIcfAexoec6NVbWuqtatXLlykNWXpKk2sIBI8tNJXjizDfwicAdwDbC+OWw9cHWzfQ3w9mY000nAYzNdUZKk4RtkF9OhwJVJZn7P56vqK0m+A1ye5BzgPuDs5vhrgdOBrcAPgXcMsG6SFslVt2znY1+9ix2PPsnhK5Zz/qnHcObxnj6cBAMLiKq6B3hlS/nfAK9vKS/g3EHVR9LC+vmgv+qW7Xzgitt58qlnANj+6JN84IrbAQyJCeBMakl//0G//dEnKWY/6K+6ZfuCj/vYV+/6+3CY8eRTz/Cxr941wNpqWAwISX1/0O949Mm9KtfSYkBI6vuD/vAVy/eqXEuLASGp7w/68089huX7L/uJsuX7L+P8U49ZtLppdAwISX1/0J95/Co+ctaxrFqxnACrViznI2cd6wnqCTHomdSSloCZD/R+hqueefwqA2FCGRDSBOpnyKof9NqdASFNGOcmaLFMTUA421PTYqEhq77ntTemIiD25RuVwaKlxrkJWixTMYqp30lA/c4ulUbJuQlaLFMREP1+o3IZAS1Fzk3QYpmKgOj3G5VNdS1Fzk3QYpmKcxDnn3rMT5yDgG7fqA5fsZztLWFgU13jziGrWgxT0YLo9xuVTXVJ02wqWhDQ3zeqfZldKklL3dQERL9sqkuaVlPRxSRJ2nu2ICQNnRNQlwYDQtJQuVbU0mEXk6ShcgLq0mFASBoqJ6AuHQaEpKFyrailw4CQNFROQF06PEktaaicgLp0GBCShs4JqEuDXUySpFYGhCSp1cADIsmyJLck+bNm/+gk30pyd5L/keSApvx5zf7W5v41g66bJGl+w2hBnAfcOWf/o8BFVbUWeAQ4pyk/B3ikql4KXNQcJ0kakYEGRJIjgDcCf9rsB3gd8KXmkE3Amc32Gc0+zf2vb46XJI3AoFsQHwfeBzzb7B8EPFpVTzf724CZoQyrgPsBmvsfa46XJI3AwAIiyb8AdlXVzXOLWw6tDvfNfd4NSbYk2fLggw8uQk0lSW0G2YI4GfiXSe4Fvkiva+njwIokM/MvjgB2NNvbgCMBmvtfBDy8+5NW1caqWldV61auXDnA6kvSdBtYQFTVB6rqiKpaA7wFuKGqfhm4EXhTc9h64Opm+5pmn+b+G6rqOS0ISdJwjGIexPuBdyfZSu8cwyVN+SXAQU35u4ELRlA3SVJjKEttVNVNwE3N9j3AiS3H/B1w9jDqM2heLUvSJHAtpkXm1bIkTQoDYpEtdLUsA6I7W2HS/Ib1/2FALDKvlrXvbIVJ8xvm/4eL9S0yr5a177xmsTS/Yf5/zNuCSHLgQg+squfMUVDvallz0x28WtbeshWmabK33UXD/P9YqIvpZnozmQOsprewXoAVwH3A0Ytemwng1bL23eErlrO95c1uK0yTpp/uomH+f8wbEFV1NECSPwauqaprm/3TgF9Y9JpMEK+WtW9shWk+kzZ4oZ9BLcP8/+hykvrVVfXOmZ2q+osk/2HRayI1bIWpzSQOXuinu2iY/x9dAuKhJB8E/ju9LqdfAf5m0WuiidTvNz5bYdrdJA4h77e7aFj/H11GMb0VWAlc2fysbMqkBc1849v+6JMUs9/4rrpl+6irpiVoEgcvnH/qMSzff9lPlI1Td+oeWxDNaKXzkrygqp4YQp20F8a5T3YSv/FpdCZx8MK4d6fuMSCS/HN6V4R7AbA6ySuBX6+qdw26clrYuPfJTuI3Po3OpA5eGOfu1C5dTBcBp9Kcd6iq7wGvGWSl1M2wJ5Rddct2Tr7wBo6+4M85+cIb9thV5KRBLaYzj1/FR846llUrlhNg1YrlfOSsY8f2w3USdFpqo6ru3+3y0M/Md6yGZ5jf0PtprUzqNz6Nzjh/255EXVoQ9zfdTJXkgCTvBe4ccL3UwTC/offTWvEbn7S0dWlBvBP4L8AqepcFvQ44d5CVUjfD/Ibeb2vFb3z7bpwHImiydQmIZ5tLhWrMDHMExCSOIFkKxn0ggiZbl4D4VpJbgUuBr3id6PEyrG/onk8YDYcKa5S6nIN4GbAReDuwNckfJHnZYKulceP5hNFwqLBGqctEuQKuB65P8lp6S268K8n3gAuq6q8GXEeNCc8nDJ9dexqlPbYgkhyU5LwkW4D3Ar8JHAy8B/j8gOsnTbVxX4pBk63LOYi/Aj4LnFlV2+aUb2mWApc0IOO+FIMmW/Z0zjlJxvXE9Lp162rLli2jroYkLSlJbq6qdXs6rksL4uAk7wP+EfD8mcKqet0+1E+SNOa6jGL6HPADepcY/ffAvcB3BlgnSdIY6BIQB1XVJcBTVfWXVfVrwEkDrpckacS6dDE91dzuTPJGYAdwxOCqJEkaB10C4j8meRG9Ya1/BPwM8O8GWitJ0sh1mSj3Z83mY8BrB1sdSdK4mDcgkvwRMO/w1qr6rYWeOMnzga8Bz2t+z5eq6kNJjga+CBwIfBd4W1X9OMnzgMuAV9G7ONGbq+revftzljZX7ZQ0ThZqQezrBIMfAa+rqieS7A98I8lfAO8GLqqqLzYT7c4BLm5uH6mqlyZ5C/BR4M37WIclw1U7JY2beQOiqjbN3U/yM73ierzLEzeT655odvdvfgp4HfBLTfkm4MP0AuKMZhvgS8Anx3mS3mJz1U5pMGyZ96/LWkzrktwO3AbckeR7SV7V5cmTLGuWCt9Fb8G/vwYeraqnm0O20bsQEc3t/QDN/Y8BB7U854YkW5JsefDBB7tUY0lw1U5p8c20zLc/+iTFbMt8T9dTV0+XeRCXAu+qqjVVdRS9q8n9ty5PXlXPVNVx9IbFngi8vO2w5jYL3Df3OTdW1bqqWrdy5cou1VgShnn5UGla9HOpXM3qEhCPV9XXZ3aq6htAp26mOY95FLiJ3gS7FUlmuraOoDevAnqtiSMBmvtfBDy8N79nKXPVTmnx2TLfN10C4ttJPp3klCQ/n+RTwE1JTkhywnwPSrIyyYpmeznwC8CdwI3Am5rD1gNXN9vXNPs0998wLecfwAvySINgy3zfdFnN9cYF7q75Fu1L8k/onYReRi+ILq+q30vyEmaHud4C/EpV/agZFvtZ4Hh6LYe3VNU9C9XN1VwlLWT30YHQa5lP+5evRVvNtar6mhxXVbfR+7Dfvfweeucjdi//O+Dsfn6XJLXxehr7pstSG5owDvvTNPFSuf0zIKaME/IkddXlJLUmiMP+JHXVZaLc2Ule2Gx/MMkVC41e0nhz2J+krrq0IH63qh5P8nPAqfRGJl082GppUBz2J6mrLgEx0x/xRuDiqroaOGBwVdIgOSFPUlddTlJvT/JpehPdPtosy+25iyXKYX+SuuoyUe4fAG8Abq+qu5McBhxbVdcNo4ILcaKcJO29RZsoBzyf3jpKJDmQ3nUeFppdLUlLmnOFeroExHfpLaL3CL0VV1cAO5PsAv5NVd08wPpJ0lA5V2hWl3MJXwFOr6qDq+og4DTgcuBdwKcGWTlpklx1y3ZOvvAGjr7gzzn5whu8JsGYcq7QrC4tiHVV9c6Znaq6LskfVNW7mxPW0lgY524Bv5UuHfsyV2ic34P96NKCeDjJ+5Mc1fy8D3gkyTLg2QHXT+pk3K8c5rfSpaPfuULj/h7sR5eA+CV6F/a5it61G1Y3ZcuAfz24qkndjfsHsDPYl45+5wqN+3uwH12W+34I+M157t66uNWR+jPuH8CHr1jO9pa6OIN9/PQ7V2jc34P92GNAJHkZ8F5gzdzj57tQkDQK4/4BfP6px7ReuMYZ7OOpnyXCx/092I8uXUz/k96V3z4InD/nRxob476EiJeUnXzj/h7sR5dRTE9XlYvzaawthSVEvHDNZFsK78G91WWpjQ8Du4Ar6c2iBqCqHh5ozTpwqQ2NyqQNZ9R0WcylNtY3t3O7lQp4ST8Vk5Y65zRoWnQZxXT0MCoiLRULDWc0IDRJuoxi2h/4t8BrmqKbgE9X1VMDrJc0tiZxOKPUpssopouBV9Fbd+lTzbYnrTW1vCqfpkWXcxCvrqpXztm/Icn3BlUhaZj6OdnsnAZNiy4B8UySn62qvwZI8hJmL0MqLVn9nmyexOGMUpsuAXE+cGOSe+hdD+Io4B0DrZU0BPtystk5DZoGXUYxbU6yFjiGXkD8oKp+tIeHSWPPk83SwuY9SZ3k1Un+IUATCMcBvwd8rLn0qLSkebJZWthCo5g+DfwYIMlrgAuBy4DHgI17euIkRya5McmdSb6f5Lym/MAk1ye5u7l9cVOeJJ9IsjXJbUlO2Nc/TlrIJK6dIy2mhQJi2ZzlNN4MbKyqL1fV7wIv7fDcTwPvqaqXAycB5yZ5BXABsLmq1gKbm33oXcp0bfOzAYfSasBcQE9a2ELnIJYl2a+qngZeT+9Du8vjAKiqncDOZvvxJHcCq4AzgFOawzbRm3j3/qb8suotDvXNJCuSHNY8jzQQnmyW5rfQB/0XgL9M8hDwJPB1gCQvpdfN1FmSNcDxwLeAQ2c+9KtqZ5JDmsNWAffPedi2psyAkKQRmDcgqur3k2wGDgOuq9llX3+K+a8w9xxJXgB8GfjtqvrbJPMe2laNlufbQNOaWb16dddqSJL20oJdRVX1zZay/931yZt1nL4MfK6qrmiKH5jpOkpyGL2lxKHXYjhyzsOPAHa0/P6NNCfJ161bt/Ba5ZKkvnVZi6kv6TUVLgHurKo/nHPXNcwuIb4euHpO+dub0UwnAY95/kGSRqfLTOp+nQy8Dbg9ya1N2e/QGy57eZJzgPuAs5v7rgVOB7YCP8TZ2pI0UgMLiKr6Bu3nFaA3Kmr34ws4d1D1kSTtnYF1MUmSljYDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQwISVKrgQVEkkuT7Epyx5yyA5Ncn+Tu5vbFTXmSfCLJ1iS3JTlhUPWSJHUzyBbEZ4A37FZ2AbC5qtYCm5t9gNOAtc3PBuDiAdZLktTBwAKiqr4GPLxb8RnApmZ7E3DmnPLLquebwIokhw2qbpKkPRv2OYhDq2onQHN7SFO+Crh/znHbmrLnSLIhyZYkWx588MGBVlaSptm4nKROS1m1HVhVG6tqXVWtW7ly5YCrJUnTa9gB8cBM11Fzu6sp3wYcOee4I4AdQ66bJGmOYQfENcD6Zns9cPWc8rc3o5lOAh6b6YqSJI3GfoN64iRfAE4BDk6yDfgQcCFweZJzgPuAs5vDrwVOB7YCPwTeMah6SZK6GVhAVNVb57nr9S3HFnDuoOoiSdp743KSWpI0ZgwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLUyoCQJLUyICRJrQwISVIrA0KS1MqAkCS1GquASPKGJHcl2ZrkglHXR5Km2dgERJJlwH8FTgNeAbw1yStGWytJml5jExDAicDWqrqnqn4MfBE4Y8R1kqSpNU4BsQq4f87+tqZMkjQC+426AnOkpayec1CyAdjQ7D6R5K69/D0HAw/t5WMmla/FLF+LWb4Wsyb1tTiqy0HjFBDbgCPn7B8B7Nj9oKraCGzs95ck2VJV6/p9/CTxtZjlazHL12LWtL8W49TF9B1gbZKjkxwAvAW4ZsR1kqSpNTYtiKp6OslvAF8FlgGXVtX3R1wtSZpaYxMQAFV1LXDtgH9N391TE8jXYpavxSxfi1lT/Vqk6jnngSVJGqtzEJKkMTJVAeFSHrOS3Jvk9iS3Jtky6voMU5JLk+xKcsecsgOTXJ/k7ub2xaOs47DM81p8OMn25r1xa5LTR1nHYUlyZJIbk9yZ5PtJzmvKp/K9AVMUEC7l0eq1VXXcFA7j+wzwht3KLgA2V9VaYHOzPw0+w3NfC4CLmvfGcc25wWnwNPCeqno5cBJwbvMZMa3vjekJCFzKQ42q+hrw8G7FZwCbmu1NwJlDrdSIzPNaTKWq2llV3222HwfupLeaw1S+N2C6AsKlPH5SAdclubmZnT7tDq2qndD7oAAOGXF9Ru03ktzWdEFNTZfKjCRrgOOBbzHF741pCohOS3lMkZOr6gR6XW7nJnnNqCuksXEx8LPAccBO4D+PtjrDleQFwJeB366qvx11fUZpmgKi01Ie06KqdjS3u4Ar6XXBTbMHkhwG0NzuGnF9RqaqHqiqZ6rqWeBPmKL3RpL96YXD56rqiqZ4at8b0xQQLuXRSPLTSV44sw38InDHwo+aeNcA65vt9cDVI6zLSM18GDb+FVPy3kgS4BLgzqr6wzl3Te17Y6omyjXD9T7O7FIevz/iKo1EkpfQazVAbzb956fptUjyBeAUeit1PgB8CLgKuBxYDdwHnF1VE3/ydp7X4hR63UsF3Av8+kwf/CRL8nPA14HbgWeb4t+hdx5i6t4bMGUBIUnqbpq6mCRJe8GAkCS1MiAkSa0MCElSKwNCktTKgJA6SvLEbvu/muSTo6qPNGgGhDRizUrD0tgxIKRFkOSoJJubBe42J1ndlH8myZvmHPdEc3tKc+2Bz9ObmCWNnbG6JrU05pYnuXXO/oHMLtfySeCyqtqU5NeAT7DnZaFPBP5xVf2fxa+qtO8MCKm7J6vquJmdJL8KzFxs6Z8BZzXbnwX+U4fn+7bhoHFmF5M0GDNr2DxN83/WLAZ3wJxj/t+wKyXtDQNCWhz/i94KwQC/DHyj2b4XeFWzfQaw/3CrJfXPgJAWx28B70hyG/A24Lym/E+An0/ybeCfYqtBS4iruUqSWtmCkCS1MiAkSa0MCElSKwNCktTKgJAktTIgJEmtDAhJUisDQpLU6v8DkcbekFDqJEAAAAAASUVORK5CYII=\n",
      "text/plain": [
       "<matplotlib.figure.Figure at 0x7fbb01feb9e8>"
      ]
     },
     "metadata": {
      "needs_background": "light"
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "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": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_log_valid = user_log.dropna(how = \"any\", subset = [\"userId\", \"sessionId\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10000"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log_valid.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "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": [
    "user_log.select(\"userId\").dropDuplicates().sort(\"userId\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_log_valid = user_log_valid.filter(user_log_valid[\"userId\"] != \"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9664"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 24,
   "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|  11|\n",
      "+------+---------+---------+------+-------------+--------+------+-----+--------------------+------+----------------+-------------+---------+----+------+-------------+--------------------+------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_log_valid.filter(\"page = 'Submit Downgrade'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "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": 25,
     "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": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "flag_downgrade_event = udf(lambda x: 1 if x == \"Submit Downgrade\" else 0, IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_log_valid = user_log_valid.withColumn(\"downgraded\", flag_downgrade_event(\"page\"))"
   ]
  },
  {
   "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', hour='22', downgraded=0)"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log_valid.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "windowval = Window.partitionBy(\"userId\").orderBy(desc(\"ts\")).rangeBetween(Window.unboundedPreceding, 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_log_valid = user_log_valid.withColumn(\"phase\", Fsum(\"downgraded\").over(windowval))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "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": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_log_valid.select([\"userId\", \"firstname\", \"ts\", \"page\", \"level\", \"phase\"]).where(user_log.userId == \"1138\").sort(\"ts\").collect()"
   ]
  },
  {
   "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}