Sparkify_full.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Customer Churn Prediction for Sparkify using PySpark\n",
    "\n",
    "This notebook is used to train the full dataset with Spark on AWS.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "0027a42b65d14b80a66f8ae98f0b6a1b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>0</td><td>application_1606254007204_0001</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-31-60.us-west-2.compute.internal:20888/proxy/application_1606254007204_0001/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-21-3.us-west-2.compute.internal:8042/node/containerlogs/container_1606254007204_0001_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Package                    Version  \n",
      "-------------------------- ---------\n",
      "beautifulsoup4             4.9.1    \n",
      "boto                       2.49.0   \n",
      "click                      7.1.2    \n",
      "jmespath                   0.10.0   \n",
      "joblib                     0.16.0   \n",
      "lxml                       4.5.2    \n",
      "mysqlclient                1.4.2    \n",
      "nltk                       3.5      \n",
      "nose                       1.3.4    \n",
      "numpy                      1.16.5   \n",
      "pip                        9.0.1    \n",
      "py-dateutil                2.2      \n",
      "python37-sagemaker-pyspark 1.4.0    \n",
      "pytz                       2020.1   \n",
      "PyYAML                     5.3.1    \n",
      "regex                      2020.7.14\n",
      "setuptools                 28.8.0   \n",
      "six                        1.13.0   \n",
      "soupsieve                  1.9.5    \n",
      "tqdm                       4.48.2   \n",
      "wheel                      0.29.0   \n",
      "windmill                   1.6"
     ]
    }
   ],
   "source": [
    "# https://aws.amazon.com/blogs/big-data/install-python-libraries-on-a-running-cluster-with-emr-notebooks/\n",
    "sc.list_packages()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "51c74984e2e840a7997a816456bed449",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pandas\n",
      "  Downloading https://files.pythonhosted.org/packages/bf/4c/cb7da76f3a5e077e545f9cf8575b8f488a4e8ad60490838f89c5cdd5bb57/pandas-1.1.4-cp37-cp37m-manylinux1_x86_64.whl (9.5MB)\n",
      "Requirement already satisfied: numpy>=1.15.4 in /usr/local/lib64/python3.7/site-packages (from pandas)\n",
      "Requirement already satisfied: pytz>=2017.2 in /usr/local/lib/python3.7/site-packages (from pandas)\n",
      "Collecting python-dateutil>=2.7.3 (from pandas)\n",
      "  Downloading https://files.pythonhosted.org/packages/d4/70/d60450c3dd48ef87586924207ae8907090de0b306af2bce5d134d78615cb/python_dateutil-2.8.1-py2.py3-none-any.whl (227kB)\n",
      "Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.7/site-packages (from python-dateutil>=2.7.3->pandas)\n",
      "Installing collected packages: python-dateutil, pandas\n",
      "Successfully installed pandas-1.1.4 python-dateutil-2.8.1\n",
      "\n",
      "Collecting scipy\n",
      "  Downloading https://files.pythonhosted.org/packages/dc/7e/8f6a79b102ca1ea928bae8998b05bf5dc24a90571db13cd119f275ba6252/scipy-1.5.4-cp37-cp37m-manylinux1_x86_64.whl (25.9MB)\n",
      "Requirement already satisfied: numpy>=1.14.5 in /usr/local/lib64/python3.7/site-packages (from scipy)\n",
      "Installing collected packages: scipy\n",
      "Successfully installed scipy-1.5.4"
     ]
    }
   ],
   "source": [
    "sc.install_pypi_package(\"pandas\")\n",
    "sc.install_pypi_package(\"scipy\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cef5de757ad7411785f2f570203aa923",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# import libraries\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from scipy.sparse import csr_matrix\n",
    "from scipy.sparse.csgraph import connected_components\n",
    "from time import time\n",
    "\n",
    "from pyspark.sql import SparkSession, Window\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "from pyspark.ml.stat import Correlation\n",
    "from pyspark.ml.feature import MinMaxScaler, VectorAssembler\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import LogisticRegression, RandomForestClassifier, GBTClassifier \n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.mllib.evaluation import MulticlassMetrics\n",
    "\n",
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "d8fde4c03efc42b7804d5479e4f7a7dd",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "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)"
     ]
    }
   ],
   "source": [
    "# load the dataset \n",
    "df = spark.read.json(\"s3n://udacity-dsnd/sparkify/sparkify_event_data.json\")\n",
    "# check the schema of the dataset\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1f648478c60e4926a43101d68ac24662",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The dataset has 26259199 rows."
     ]
    }
   ],
   "source": [
    "print('The dataset has {} rows.'.format(df.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "269c6c68dd8d4ef485c7fe363159ccfd",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|         Start time|\n",
      "+-------------------+\n",
      "|2018-10-01 00:00:01|\n",
      "+-------------------+"
     ]
    }
   ],
   "source": [
    "df.select(min(to_timestamp(col('ts')/1000)).alias('Start time')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a359c867ef2d44b985281df3af134273",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|           End time|\n",
      "+-------------------+\n",
      "|2018-12-01 00:00:02|\n",
      "+-------------------+"
     ]
    }
   ],
   "source": [
    "df.select(max(to_timestamp(col('ts')/1000)).alias('End time')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Cleaning\n",
    "\n",
    "One thing to note here is that unlike the mini dataset, unregistered users with no information on the first name, last name, gender, location, registration date and user agent do have a user ID in the full dataset (all of these users were assigned the user ID 1261737). Since we need to calculate each user's registration duration as one of the featues, we remove the records when **registration** is null."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9d6181c7121a4a528d3bfa3f5a37857a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def clean_data(df):\n",
    "    \"\"\"Clean a Sparkify dataset \n",
    "    \n",
    "    Args:\n",
    "    df: (spark dataframe) a Sparkify dataset\n",
    "    \n",
    "    Returns:\n",
    "    df: (spark dataframe) a preprocessed Sparkify dataset\n",
    "    \"\"\"\n",
    "    # remove records when 'registration' is null\n",
    "    df = df.filter(df['registration'].isNotNull())\n",
    "    \n",
    "    # convert 'registration' and 'ts' to date format\n",
    "    df = df \\\n",
    "        .withColumn('registrationTime', to_timestamp(col('registration')/1000)) \\\n",
    "        .withColumn('time', to_timestamp(col('ts')/1000)) \n",
    "    \n",
    "    # replace location with first listed state \n",
    "    state_udf = udf(lambda x: x.split(', ')[1].split('-')[0])\n",
    "    df = df.withColumn('location', state_udf('location'))\n",
    "        \n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Labeling"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "49a595461ae243fa874d7d2caef1a463",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def label_data(df, label='Churn'):\n",
    "    \"\"\"Add a label column to the Sparkify dataset \n",
    "    \n",
    "    Args:\n",
    "    df: (spark dataframe) a cleaned Sparkify dataset\n",
    "    label: (string) label name\n",
    "    \n",
    "    Returns:\n",
    "    df: (spark dataframe) a labeled Sparkify dataset\n",
    "    \"\"\"\n",
    "    userWindow = Window.partitionBy('userId').orderBy('ts').rangeBetween(Window.unboundedPreceding,Window.unboundedFollowing)\n",
    "\n",
    "    # label churned users to be 1 and unchurned users to be 0 \n",
    "    df = df \\\n",
    "        .withColumn(label, when(col('page')=='Cancellation Confirmation', 1).otherwise(0)) \\\n",
    "        .withColumn(label, max(label).over(userWindow))\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Engineering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cdedd42b738549b98a4c90f2bef9448b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def build_features(df, label='Churn'):\n",
    "    \"\"\"Build features to be used for modeling\n",
    "    \n",
    "    Args:\n",
    "    df: (spark dataframe) a cleaned and labeled Sparkify dataset\n",
    "    label: (string) label name\n",
    "    \n",
    "    Returns:\n",
    "    user_df: (spark dataframe) a labeled dataset with features of interest grouped by user ids\n",
    "    \"\"\"\n",
    "    userWindow = Window.partitionBy('userId').orderBy('ts').rangeBetween(Window.unboundedPreceding,Window.unboundedFollowing)\n",
    "    \n",
    "    # calculate the duration between registration to last activity (in days)\n",
    "    regist_duration_df = df.groupBy('userId') \\\n",
    "        .agg(((last(col('ts'))-last(col('registration')))/1000/3600/24).alias('registDuration'))\n",
    "\n",
    "    # compute average session duration (in hours)\n",
    "    avg_session_duration_df = df \\\n",
    "        .groupBy(['userId', 'sessionId']).agg(min(col('ts')).alias('session_start'), max(col('ts')).alias('session_end'))\\\n",
    "        .groupBy('userId').agg(avg((col('session_end') - col('session_start'))/1000/3600).alias('avgSessionDuration'))\n",
    "\n",
    "    # define the default start and end of the observation period\n",
    "    obs_start_default = df.select(min(col('ts'))).collect()[0][0]\n",
    "    obs_end_default = df.select(max(col('ts'))).collect()[0][0]\n",
    "\n",
    "    # compute the observation period\n",
    "    df = df \\\n",
    "        .withColumn('obs_start', when(col('registration') > obs_start_default, first(col('ts')).over(userWindow)) \\\n",
    "                    .otherwise(obs_start_default)) \\\n",
    "        .withColumn('end_state', last(col('page')).over(userWindow)) \\\n",
    "        .withColumn('obs_end', when(col('end_state') == 'Cancellation Confirmation', last(col('ts')).over(userWindow)) \\\n",
    "                    .otherwise(obs_end_default)) \\\n",
    "        .withColumn('obsDays', (col('obs_end') - col('obs_start'))/1000/3600/24)\n",
    "\n",
    "    # aggregate activity statistics\n",
    "    user_df = df.groupBy('userId') \\\n",
    "        .agg(first(col(label)).alias(label), \\\n",
    "             first(col('obsDays')).alias('obsDays'), \\\n",
    "             sum(when(col('page') == 'NextSong', 1).otherwise(0)).alias('nSongs'), \\\n",
    "             sum(when(col('page') == 'Thumbs Up', 1).otherwise(0)).alias('nThumbsUp'), \\\n",
    "             sum(when(col('page') == 'Thumbs Down', 1).otherwise(0)).alias('nThumbsDown'), \\\n",
    "             sum(when((col('page') == 'Upgrade') | (col('page') == 'Submit Upgrade'), 1) \\\n",
    "                 .otherwise(0)).alias('nUpgrade'), \\\n",
    "             sum(when((col('page') == 'Downgrade') | (col('page') == 'Submit Downgrade'), 1) \\\n",
    "                 .otherwise(0)).alias('nDowngrade'), \\\n",
    "             sum(when(col('page') == 'Add Friend', 1).otherwise(0)).alias('nAddFriend'), \\\n",
    "             sum(when(col('page') == 'Add to Playlist', 1).otherwise(0)).alias(\"nAddPlaylist\"), \\\n",
    "             sum(when(col('page') == 'Roll Advert', 1).otherwise(0)).alias('nAdvert'), \\\n",
    "             sum(when((col('page') == 'Help'), 1).otherwise(0)).alias('nHelp'), \\\n",
    "             sum(when((col('page') == 'Error'), 1).otherwise(0)).alias('nError')) \\\n",
    "        .join(regist_duration_df, on='userId') \\\n",
    "        .join(avg_session_duration_df, on='userId')\n",
    "\n",
    "    user_df = user_df \\\n",
    "        .withColumn('avgDailySongs', col('nSongs') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyThumbsUp', col('nThumbsUp') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyThumbsDown', col('nThumbsDown') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyUpgrade', col('nUpgrade') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyDowngrade', col('nDowngrade') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyAddFriend', col('nAddFriend') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyAddPlaylist', col('nAddPlaylist') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyAdvert', col('nAdvert') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyHelp', col('nHelp') / col('obsDays')) \\\n",
    "        .withColumn('avgDailyError', col('nError') / col('obsDays')) \\\n",
    "        .drop('userId', 'obsDays', 'nSongs', 'nThumbsUp', 'nThumbsDown', 'nUpgrade', 'nDowngrade', \\\n",
    "              'nAddFriend', 'nAddPlaylist', 'nAdvert', 'nHelp', 'nError')\n",
    "    \n",
    "    return user_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "19366445a78c4d34aafb28a930aec01a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def drop_multicollinear_features(user_df, label='Churn', threshold=0.85):\n",
    "    \"\"\"Drop highly correlated features to avoid multicollinearity\n",
    "    \n",
    "    Args:\n",
    "    user_df: (spark dataframe) a labeled dataset with binary and numerical features\n",
    "    label: (string) label name\n",
    "    threshold: (float) the threshold of high correlation\n",
    "    \n",
    "    Returns:\n",
    "    model_df: (spark dataframe) a labeled dataset after removal of multicollinear features\n",
    "    \"\"\"\n",
    "    vec_col = 'corr_features'\n",
    "   \n",
    "    # assemble all vector columns into one vector column\n",
    "    assembler = VectorAssembler(inputCols=user_df.columns, outputCol=vec_col)\n",
    "    corr_df = assembler.transform(user_df).select(vec_col)\n",
    "\n",
    "    # compute the correlation between 'churn' and every feature and the correlation between each pair of features\n",
    "    corr_mat = Correlation.corr(corr_df, vec_col)\n",
    "    # convert the corrlation matrix to a pandas dataframe with column names\n",
    "    corr_values = corr_mat.collect()[0][0].values\n",
    "    corr_mat_pd = pd.DataFrame(corr_values.reshape(-1, len(user_df.columns)), \\\n",
    "                           index=user_df.columns, columns=user_df.columns)\n",
    "    \n",
    "    # construct an adjacency matrix where high correlation is labeled as 1, otherwise 0\n",
    "    is_high_corr = np.abs(corr_mat_pd.values) > threshold\n",
    "    adj_mat = csr_matrix(is_high_corr.astype(int) - np.identity(len(user_df.columns)))\n",
    "\n",
    "    # find groups of highly correlated features by finding the connected components in the adjacency matrix\n",
    "    _, corr_labels = connected_components(csgraph=adj_mat, directed=False)\n",
    "    unique, unique_counts = np.unique(corr_labels, return_counts=True)\n",
    "    # get groups with size > 1\n",
    "    high_corr_labels = unique[unique_counts > 1]\n",
    "\n",
    "    # if there is at least one group of highly correlated features\n",
    "    if len(high_corr_labels) > 0:\n",
    "        # map the label indices of highly correlated features to their column names\n",
    "        print('Highly correlated features include:')\n",
    "        high_corr_col_dict = {}\n",
    "        for high_corr_label in high_corr_labels:\n",
    "            high_corr_col_dict[high_corr_label] = [col_name for corr_label, col_name in zip(corr_labels, user_df.columns) \n",
    "                                               if corr_label == high_corr_label]\n",
    "            print(high_corr_col_dict[high_corr_label])\n",
    "        \n",
    "        print('\\nFeatures to keep:')\n",
    "        cols_to_drop = []\n",
    "        for col_name_list in high_corr_col_dict.values(): \n",
    "            # keep the feature that has the highest correlation with the response variable\n",
    "            col_to_keep = corr_mat_pd.loc[col_name_list, label].idxmax()\n",
    "            print(col_to_keep)\n",
    "            # remove the other features to avoid multicolinearity \n",
    "            col_name_list.remove(col_to_keep)\n",
    "            corr_mat_pd.drop(index=col_name_list, columns=col_name_list, inplace=True)\n",
    "            cols_to_drop.extend(col_name_list)\n",
    "            \n",
    "    model_df = user_df.drop(*cols_to_drop)\n",
    "    \n",
    "    return model_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Modeling and Evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8a0f595f24484ce8b2af9dd069607b0a",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def split_train_test(model_df, fraction, label='Churn', seed=2020):\n",
    "    \"\"\"Split the dataset into a training and a test set using stratified sampling based on the label column\n",
    "    \n",
    "    Args:\n",
    "    model_df: (spark dataframe) a labeled dataset with binary and numerical features\n",
    "    fraction: (float) the fraction of the dataset used for training\n",
    "    label: (string) label name\n",
    "    seed: (int) a nonnegative integer \n",
    "    \n",
    "    Returns:\n",
    "    train: (spark dataframe) a training set\n",
    "    test: (spark dataframe) a test set\n",
    "    \"\"\"\n",
    "    train = model_df.sampleBy(label, fractions={0: fraction, 1: fraction}, seed=seed)\n",
    "    test = model_df.subtract(train)\n",
    "    \n",
    "    return train, test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c7c002a1d4c844bcbebc3eabf8a08fbb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def print_metrics(pred, label='Churn'):\n",
    "    \"\"\"Print evaluation metrics on a test set\n",
    "    \n",
    "    Args:\n",
    "    pred: (spark dataframe) a test set \n",
    "    \n",
    "    Returns:\n",
    "    summary: (pandas dataframe) a summary of evaluation metrics\n",
    "    \"\"\"\n",
    "    eval_metrics = {}\n",
    "\n",
    "    # compute area under PR curve\n",
    "    evaluator = BinaryClassificationEvaluator(labelCol=label)\n",
    "    auc_pr = evaluator.evaluate(pred, {evaluator.metricName:'areaUnderPR'})\n",
    "\n",
    "    # compute precision, recall and f1 score\n",
    "    predictionAndLabels = pred.select('prediction', label)\n",
    "    # both 'prediction' and label in predictionAndLabels need to be cast to float type and \n",
    "    # map to tuple before calling 'MulticlassMetrics'\n",
    "    metrics = MulticlassMetrics(predictionAndLabels.rdd.map(lambda x: tuple(map(float, x))))\n",
    "\n",
    "    # get overall statistics\n",
    "    eval_metrics['overall'] = [metrics.weightedPrecision, metrics.weightedRecall, \\\n",
    "                               metrics.weightedFMeasure(), auc_pr]\n",
    "\n",
    "    # get statistics by class\n",
    "    classes = [0.0, 1.0]\n",
    "    for cls in classes:\n",
    "        eval_metrics['class ' + str(int(cls))] = [metrics.precision(cls), metrics.recall(cls), \\\n",
    "                                                  metrics.fMeasure(cls), '']\n",
    "\n",
    "    # convert to a pandas dataframe for display\n",
    "    summary = pd.DataFrame.from_dict(eval_metrics, orient='index', \\\n",
    "                                     columns=['precision', 'recall', 'f1 score', 'AUC-PR'])   \n",
    "    \n",
    "    return summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9fec43643db6462fbfb1caf2f54a50ec",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def build_pipeline(bin_cols, num_cols, label='Churn', seed=2020):\n",
    "    \"\"\"Build a pipeline using a random forest classifier for training\n",
    "    \n",
    "    Args:\n",
    "    bin_cols: (list) a list of binary columns\n",
    "    num_cols: (list) a list of numerical columns\n",
    "    label: (string) label name\n",
    "    seed: (int) a nonnegative integer \n",
    "    \n",
    "    Returns:\n",
    "    pipeline_rf: (Pipeline object) a sequence of stages with the last stage to be a random forest classifier \n",
    "    \"\"\"\n",
    "    # assemble numerical columns to a single vector column \n",
    "    num_assembler = VectorAssembler(inputCols=num_cols, outputCol='num_features')\n",
    "    \n",
    "    # scale each numberical feature within the range [0,1] \n",
    "    scaler = MinMaxScaler(inputCol='num_features', outputCol='scaled_features')\n",
    "    \n",
    "    # assemble all vector columns into one vector column\n",
    "    assembler = VectorAssembler(inputCols=bin_cols + ['scaled_features'], outputCol='features')\n",
    "\n",
    "    # random forest classifier\n",
    "    rf = RandomForestClassifier(featuresCol='features', labelCol=label, seed=seed)\n",
    "    pipeline_rf = Pipeline(stages=[num_assembler, scaler, assembler, rf])\n",
    "    \n",
    "    return pipeline_rf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a0e3d53d23034ddaa910bbf1477e7aa7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def tune_rf(train, pipeline_rf, numTrees=[100, 200], maxDepth=[4, 5], label='Churn'):\n",
    "    \"\"\"Tune the hyperameters of the random forest classifier using grid search with cross validation\n",
    "    \n",
    "    Args:\n",
    "    train: (spark dataframe) a training set\n",
    "    pipeline_rf: (Pipeline object) a sequence of stages with the last stage to be a random forest classifier \n",
    "    numTrees: (list) number of trees\n",
    "    maxDepth: (list) maximum tree depth\n",
    "    \n",
    "    Returns:\n",
    "    cv_rf: (CrossValidator object) a cross validation model trained by the random forest classifier\n",
    "    \"\"\"\n",
    "    # set hyperparameters for tuning\n",
    "    paramGrid = ParamGridBuilder() \\\n",
    "                .addGrid(pipeline_rf.getStages()[-1].numTrees, numTrees) \\\n",
    "                .addGrid(pipeline_rf.getStages()[-1].maxDepth, maxDepth) \\\n",
    "                .build()  \n",
    "\n",
    "    # grid search with cross validation    \n",
    "    crossval_rf = CrossValidator(estimator = pipeline_rf,\n",
    "                                 estimatorParamMaps = paramGrid,\n",
    "                                 evaluator = BinaryClassificationEvaluator(labelCol=label, metricName='areaUnderPR'),\n",
    "                                 numFolds = 4)\n",
    "\n",
    "    start = time()\n",
    "    cv_rf = crossval_rf.fit(train)\n",
    "    end = time()\n",
    "    print('Total training time for hyperparameter tuning on random forest classifier: {:.0f} seconds'.format(end - start))\n",
    "    \n",
    "    return cv_rf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## End to End Workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "6468fb47f54b40f2a3358fbc51167c70",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|Churn|count|\n",
      "+-----+-----+\n",
      "|    1| 5003|\n",
      "|    0|17274|\n",
      "+-----+-----+"
     ]
    }
   ],
   "source": [
    "df = clean_data(df)\n",
    "df = label_data(df)\n",
    "df.dropDuplicates(['userId']).groupby('Churn').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "dece4a35355e4222ac82f934307570ed",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Highly correlated features include:\n",
      "['avgDailySongs', 'avgDailyThumbsUp', 'avgDailyAddPlaylist']\n",
      "\n",
      "Features to keep:\n",
      "avgDailySongs\n",
      "\n",
      "The schema of the model for training:\n",
      "root\n",
      " |-- Churn: integer (nullable = true)\n",
      " |-- registDuration: double (nullable = true)\n",
      " |-- avgSessionDuration: double (nullable = true)\n",
      " |-- avgDailySongs: double (nullable = true)\n",
      " |-- avgDailyThumbsDown: double (nullable = true)\n",
      " |-- avgDailyUpgrade: double (nullable = true)\n",
      " |-- avgDailyDowngrade: double (nullable = true)\n",
      " |-- avgDailyAddFriend: double (nullable = true)\n",
      " |-- avgDailyAdvert: double (nullable = true)\n",
      " |-- avgDailyHelp: double (nullable = true)\n",
      " |-- avgDailyError: double (nullable = true)"
     ]
    }
   ],
   "source": [
    "user_df = build_features(df)\n",
    "model_df = drop_multicollinear_features(user_df)\n",
    "print('\\nThe schema of the model for training:')\n",
    "model_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3289200ca5344d96a7d92c7a062be1a5",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Exception in thread cell_monitor-31:\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/conda/lib/python3.7/threading.py\", line 926, in _bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/opt/conda/lib/python3.7/threading.py\", line 870, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/opt/conda/lib/python3.7/site-packages/awseditorssparkmonitoringwidget-1.0-py3.7.egg/awseditorssparkmonitoringwidget/cellmonitor.py\", line 178, in cell_monitor\n",
      "    job_binned_stages[job_id][stage_id] = all_stages[stage_id]\n",
      "KeyError: 6490\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total training time for hyperparameter tuning on random forest classifier: 3997 seconds"
     ]
    }
   ],
   "source": [
    "train, test = split_train_test(model_df, fraction=0.8)\n",
    "num_cols = [field.name for field in model_df.schema.fields if field.dataType != IntegerType()]\n",
    "bin_cols = [col for col in model_df.columns if col not in num_cols + ['Churn']]\n",
    "pipeline_rf = build_pipeline(bin_cols, num_cols)\n",
    "cv_rf = tune_rf(train, pipeline_rf, maxDepth=[8, 10])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "36fe2433aa124cb28652bf27f2e6fdc8",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "   numTrees  maxDepth    AUC-PR\n",
      "0       100         8  0.829046\n",
      "1       100        10  0.860917\n",
      "2       200         8  0.829139\n",
      "3       200        10  0.859938"
     ]
    }
   ],
   "source": [
    "# store grid search results in a dataframe\n",
    "params = [{p.name: v for p, v in m.items()} for m in cv_rf.getEstimatorParamMaps()]\n",
    "params_pd = pd.DataFrame(params)\n",
    "params_pd['AUC-PR'] = cv_rf.avgMetrics\n",
    "params_pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "de53a116d4cf4697a619ccc215b40f99",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Exception in thread cell_monitor-33:\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/conda/lib/python3.7/threading.py\", line 926, in _bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/opt/conda/lib/python3.7/threading.py\", line 870, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/opt/conda/lib/python3.7/site-packages/awseditorssparkmonitoringwidget-1.0-py3.7.egg/awseditorssparkmonitoringwidget/cellmonitor.py\", line 178, in cell_monitor\n",
      "    job_binned_stages[job_id][stage_id] = all_stages[stage_id]\n",
      "KeyError: 8595\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "         precision    recall  f1 score    AUC-PR\n",
      "overall   0.915436  0.914031  0.907809  0.891642\n",
      "class 0   0.910937  0.987454  0.947654          \n",
      "class 1   0.932165  0.641026  0.759656"
     ]
    }
   ],
   "source": [
    "test_prediction = cv_rf.transform(test)\n",
    "print_metrics(test_prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "86901ac4b08a47ca8d1dc86365bcbb43",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "              feature  importance\n",
      "7      avgDailyAdvert    0.188268\n",
      "4     avgDailyUpgrade    0.174162\n",
      "0      registDuration    0.170745\n",
      "9       avgDailyError    0.122517\n",
      "3  avgDailyThumbsDown    0.090220\n",
      "5   avgDailyDowngrade    0.089500\n",
      "8        avgDailyHelp    0.046742\n",
      "2       avgDailySongs    0.045637\n",
      "6   avgDailyAddFriend    0.037032\n",
      "1  avgSessionDuration    0.035177"
     ]
    }
   ],
   "source": [
    "features = bin_cols + num_cols\n",
    "importances = list(cv_rf.bestModel.stages[-1].featureImportances)\n",
    "feat_imp_pd = pd.DataFrame({'feature': features, 'importance': importances}).sort_values('importance', ascending = False)\n",
    "feat_imp_pd"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}