Sparkify.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sparkify Project Workspace\n",
    "This workspace contains a tiny subset (128MB) of the full dataset available (12GB). Feel free to use this workspace to build your project, or to explore a smaller subset with Spark before deploying your cluster on the cloud. Instructions for setting up your Spark cluster is included in the last lesson of the Extracurricular Spark Course content.\n",
    "\n",
    "You can follow the steps below to guide your data analysis and model building portion of this project."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import libraries\n",
    "# 1) import any other libraries you might need\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import avg, desc, min, max, split, udf\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.feature import CountVectorizer, StandardScaler, StringIndexer, VectorAssembler, OneHotEncoder\n",
    "from pyspark.ml.regression import LinearRegression\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a Spark session\n",
    "spark = SparkSession.builder\\\n",
    "                    .appName('Sparkify')\\\n",
    "                    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load and Clean Dataset\n",
    "In this workspace, the mini-dataset file is `mini_sparkify_event_data.json`. Load and clean the dataset, checking for invalid or missing data - for example, records without userids or sessionids. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# I cannot upload the dataset in github so if you want to download or use it you can use this link: \n",
    "# https://video.udacity-data.com/topher/2018/December/5c1d6681_medium-sparkify-event-data/medium-sparkify-event-data.json\n",
    "df = spark.read.json(\"mini_sparkify_event_data.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploratory Data Analysis\n",
    "When you're working with the full dataset, perform EDA by loading a small subset of the data and doing basic manipulations within Spark. In this workspace, you are already provided a small subset of data you can explore.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(artist='Martha Tilston', auth='Logged In', firstName='Colin', gender='M', itemInSession=50, lastName='Freeman', length=277.89016, level='paid', location='Bakersfield, CA', method='PUT', page='NextSong', registration=1538173362000, sessionId=29, song='Rockpools', status=200, ts=1538352117000, userAgent='Mozilla/5.0 (Windows NT 6.1; WOW64; rv:31.0) Gecko/20100101 Firefox/31.0', userId='30')"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|    30|\n",
      "|     9|\n",
      "|    30|\n",
      "|     9|\n",
      "|    30|\n",
      "|     9|\n",
      "|     9|\n",
      "|    30|\n",
      "|    30|\n",
      "|    30|\n",
      "|     9|\n",
      "|     9|\n",
      "|    30|\n",
      "|     9|\n",
      "|     9|\n",
      "|    30|\n",
      "|     9|\n",
      "|    74|\n",
      "|    30|\n",
      "|     9|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('userId').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deleting any useless rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.where(\"userId != '' and userId != ' '\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.dropna(how='any', subset=['userId', 'sessionId'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "valid = df.dropDuplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|    10|\n",
      "|   100|\n",
      "|100001|\n",
      "|100002|\n",
      "|100003|\n",
      "|100004|\n",
      "|100005|\n",
      "|100006|\n",
      "|100007|\n",
      "|100008|\n",
      "|100009|\n",
      "|100010|\n",
      "|100011|\n",
      "|100012|\n",
      "|100013|\n",
      "|100014|\n",
      "|100015|\n",
      "|100016|\n",
      "|100017|\n",
      "|100018|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.groupby('userId').max().select('userId').sort('userId').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define Churn\n",
    "\n",
    "Once you've done some preliminary analysis, create a column `Churn` to use as the label for your model. I suggest using the `Cancellation Confirmation` events to define your churn, which happen for both paid and free users. As a bonus task, you can also look into the `Downgrade` events.\n",
    "\n",
    "### Explore Data\n",
    "Once you've defined churn, perform some exploratory data analysis to observe the behavior for users who stayed vs users who churned. You can start by exploring aggregates on these two groups of users, observing how much of a specific action they experienced per a certain time unit or number of songs played."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                page|\n",
      "+--------------------+\n",
      "|               About|\n",
      "|          Add Friend|\n",
      "|     Add to Playlist|\n",
      "|              Cancel|\n",
      "|Cancellation Conf...|\n",
      "|           Downgrade|\n",
      "|               Error|\n",
      "|                Help|\n",
      "|                Home|\n",
      "|              Logout|\n",
      "|            NextSong|\n",
      "|         Roll Advert|\n",
      "|       Save Settings|\n",
      "|            Settings|\n",
      "|    Submit Downgrade|\n",
      "|      Submit Upgrade|\n",
      "|         Thumbs Down|\n",
      "|           Thumbs Up|\n",
      "|             Upgrade|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.select('page').dropDuplicates().sort('page').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have the following values that are in the `page` column.\n",
    "\n",
    "* Add Friend\n",
    "* Add to Playlist\n",
    "* Cancel\n",
    "* Cancellation Confirmation\n",
    "* Downgrade\n",
    "* Error\n",
    "* Help\n",
    "* Home\n",
    "* Logout\n",
    "* NextSong\n",
    "* Roll Advert\n",
    "* Save Settings\n",
    "* Settings\n",
    "* Submit Downgrade\n",
    "* Submit Upgrade\n",
    "* Thumbs Down\n",
    "* Thumbs Up\n",
    "* Upgrade<br>\n",
    "The pages we never want to see users utilize is the **Cancel** and especially the `Cancellation Confirmation`. </p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "churn = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "valid = valid.withColumn('churn', churn(valid.page))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "52"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid.where(\"churn == 1\").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Feature Engineering\n",
    "Once you've familiarized yourself with the data, build out the features you find promising to train your model on. To work with the full dataset, you can follow the following steps.\n",
    "- Write a script to extract the necessary features from the smaller subset of data\n",
    "- Ensure that your script is scalable, using the best practices discussed in Lesson 3\n",
    "- Try your script on the full data set, debugging your script if necessary\n",
    "\n",
    "If you are working in the classroom workspace, you can just extract features based on the small subset of data contained here. Be sure to transfer over this work to the larger dataset when you work on your Spark cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+------+\n",
      "|churn|gender| count|\n",
      "+-----+------+------+\n",
      "|    1|     F|    20|\n",
      "|    0|     M|123544|\n",
      "|    1|     M|    32|\n",
      "|    0|     F|154558|\n",
      "+-----+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.groupby([\"churn\", 'gender']).count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "gender_idx = StringIndexer(inputCol=\"gender\", outputCol='gender_idx')\n",
    "page_idx = StringIndexer(inputCol=\"page\", outputCol='page_idx')\n",
    "gender_encoder = OneHotEncoder(inputCol='gender_idx', outputCol='gender_vec')\n",
    "page_encoder = OneHotEncoder(inputCol='page_idx', outputCol='page_vec')\n",
    "assembler = VectorAssembler(inputCols=[\"gender_vec\", \"page_vec\", \"status\"], outputCol=\"features\")\n",
    "idx = StringIndexer(inputCol=\"churn\", outputCol=\"label\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Modeling\n",
    "Split the full dataset into train, test, and validation sets. Test out several of the machine learning methods you learned. Evaluate the accuracy of the various models, tuning parameters as necessary. Determine your winning model based on test accuracy and report results on the validation set. Since the churned users are a fairly small subset, I suggest using F1 score as the metric to optimize."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "model =  LogisticRegression(maxIter=15)\n",
    "pipeline = Pipeline(stages=[gender_idx, page_idx, gender_encoder, page_encoder, assembler, idx, model])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> I chose `LogisticRegression` as we want a simple `1` or `0` prediction <br>\n",
    "> now let's split our data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "rest, validation = valid.randomSplit([0.9, 0.1], seed=42)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> Building the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "param = ParamGridBuilder() \\\n",
    "    .addGrid(model.regParam,[0.0, 0.1, ]) \\\n",
    "    .addGrid(model.elasticNetParam,[0.0, 0.1, 0.5, 1.0]) \\\n",
    "    .build()\n",
    "\n",
    "\n",
    "cross = CrossValidator(estimator=pipeline,\n",
    "                          estimatorParamMaps=param,\n",
    "                          evaluator=MulticlassClassificationEvaluator(),\n",
    "                          numFolds=3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fit the Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "fitted = cross.fit(rest)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = fitted.transform(validation)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+----------+\n",
      "|       rawPrediction|         probability|prediction|\n",
      "+--------------------+--------------------+----------+\n",
      "|[21.6661123683481...|[0.99999999961048...|       0.0|\n",
      "|[21.6640372010717...|[0.99999999960967...|       0.0|\n",
      "|[21.6657777009774...|[0.99999999961035...|       0.0|\n",
      "|[21.6640372010717...|[0.99999999960967...|       0.0|\n",
      "|[17.7169719283065...|[0.99999997978764...|       0.0|\n",
      "|[17.3379741456721...|[0.99999997047339...|       0.0|\n",
      "|[21.2936050554995...|[0.99999999943466...|       0.0|\n",
      "|[21.2936050554995...|[0.99999999943466...|       0.0|\n",
      "|[21.6640372010717...|[0.99999999960967...|       0.0|\n",
      "|[17.7107117119453...|[0.99999997966071...|       0.0|\n",
      "|[17.3465397827343...|[0.99999997072523...|       0.0|\n",
      "|[17.3465397827343...|[0.99999997072523...|       0.0|\n",
      "|[18.0933973655363...|[0.99999998612805...|       0.0|\n",
      "|[18.4638295111085...|[0.99999999042231...|       0.0|\n",
      "|[18.0933973655363...|[0.99999998612805...|       0.0|\n",
      "|[18.4638295111085...|[0.99999999042231...|       0.0|\n",
      "|[18.4638295111085...|[0.99999999042231...|       0.0|\n",
      "|[18.0933973655363...|[0.99999998612805...|       0.0|\n",
      "|[18.0933973655363...|[0.99999998612805...|       0.0|\n",
      "|[18.0933973655363...|[0.99999998612805...|       0.0|\n",
      "+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "results.select(\"rawPrediction\",\"probability\", \"prediction\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1.0,\n",
       " 1.0,\n",
       " 1.0,\n",
       " 1.0,\n",
       " 0.9996942448433899,\n",
       " 0.9996942448433899,\n",
       " 0.9996942448433899,\n",
       " 0.9996942448433899]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fitted.avgMetrics"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As seen, we have an awesome score accuracy about 0.99 with 15 iterations!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can improve our model more by tuning new features and trying different machine learning algorithm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Final Steps\n",
    "Clean up your code, adding comments and renaming variables to make the code easier to read and maintain. Refer to the Spark Project Overview page and Data Scientist Capstone Project Rubric to make sure you are including all components of the capstone project and meet all expectations. Remember, this includes thorough documentation in a README file in a Github repository, as well as a web app or blog post."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Done**"
   ]
  }
 ],
 "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
}