Technology/Artificial Intelligence/Prediction/Customer_Churn_Prediction_Using_PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Customer Churn Prediction of a Music App using PySpark\n",
    "![](banner.jpeg)\n",
    "\n",
    "### Table of Content\n",
    "- Introduction\n",
    "- Problem Statement\n",
    "- Import libraries\n",
    "- Create PySpark Session\n",
    "- Load Dataset\n",
    "- Exploratory Data Analysis\n",
    "    - Define Churn\n",
    "    - Explore Data\n",
    "    - Filter Data\n",
    "- Feature Engineering\n",
    "    - Create new features\n",
    "    - Scale features\n",
    "        - VectorAssembler()\n",
    "        - MinMaxScaler()\n",
    "- Modeling\n",
    "    - Train/Test Split\n",
    "- Machine Learning Classifier\n",
    "    - Train (Logistic Regression)\n",
    "        - Summary\n",
    "        - Area Under ROC\n",
    "    - Prediction on validation set\n",
    "    - Evaluation\n",
    "        - Accuracy\n",
    "        - Area Under ROC\n",
    "    - Tune Hyperparameter\n",
    "        - ParamGridBuilder()\n",
    "        - CrossValidator()\n",
    "    - Evaluation (Tuned Model)\n",
    "        - Accuracy\n",
    "        - Area Under ROC\n",
    "    - Comparison\n",
    "- Summary\n",
    "- Troubleshooting\n",
    "- Credits\n",
    "    \n",
    "Post your doubt/feedback/discussion in our FB group unit [here](https://www.facebook.com/groups/colearninglounge/learning_content/?filter=387225811930602) in the appropriate section.        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What is customer churn?\n",
    "Customer churn is the percentage of customers that stopped using your company’s product or service during a certain time frame. You can ascertain churn rate by dividing the number of customers you lost during that timespan — say a quarter — by the number of customers you had toward the start of that timeframe.\n",
    "\n",
    "For instance, you begin your quarter with 400 customers and end with 380, your churn rate is 5% in light of the fact that you lost 5% of your clients.\n",
    "\n",
    "\n",
    "### Why is the churn rate important?\n",
    "You may be wondering why it’s necessary to calculate churn rate. Naturally, you’re going to lose some customers here and there, and 5% doesn’t sound too bad, right?\n",
    "\n",
    "Well, it’s important because it costs more to acquire new customers than it does to retain existing customers. In fact, an increase in customer retention of just 5% can create at least a 25% increase in profit. This is because returning customers will likely spend 67% more on your company’s products and services. As a result, your company can spend less on the operating costs of having to acquire new customers. You don’t need to spend time and money on convincing an existing customer to select your company over competitors because they’ve already made that decision."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Problem Statement\n",
    "The dataset contains two months of user's behavior log of a music app. The log contains some basic information about the user as well as information about a single action. A user can contain many entries. In the data, a part of the user is churned, through the cancellation of the account behavior can be distinguished.\n",
    "\n",
    "The job of the project is to find the characteristics of churned users from the behavioral data of these users, and take measures to retain the potential lost users as early as possible."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import avg, col, concat, desc, explode, lit, min, max, split, udf, count, when, isnull, collect_list\n",
    "from pyspark.sql.types import IntegerType, BooleanType, FloatType\n",
    "from pyspark.ml.feature import MinMaxScaler\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create PySpark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"MusicApp\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load Dataset\n",
    "For this project, you can download the public dataset from [Kaggle](https://www.kaggle.com/rowhitswami/music-app-logs). 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": {
    "scrolled": true
   },
   "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": [
    "data_path = 'data.json'\n",
    "df = spark.read.json(data_path)\n",
    "# See the frame schema\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "286500"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Total number of records in our dataset\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+\n",
      "|artist|auth|firstName|gender|itemInSession|lastName|length|level|location|method|page|registration|sessionId| song|status| ts|userAgent|userId|\n",
      "+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+\n",
      "| 58392|   0|     8346|  8346|            0|    8346| 58392|    0|    8346|     0|   0|        8346|        0|58392|     0|  0|     8346|     0|\n",
      "+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select([count(when(isnull(x), x)).alias(x) for x in df.columns]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      auth|\n",
      "+----------+\n",
      "|Logged Out|\n",
      "| Cancelled|\n",
      "|     Guest|\n",
      "| Logged In|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('auth').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|level|\n",
      "+-----+\n",
      "| free|\n",
      "| paid|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('level').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|            location|\n",
      "+--------------------+\n",
      "|     Gainesville, FL|\n",
      "|Atlantic City-Ham...|\n",
      "|Deltona-Daytona B...|\n",
      "|San Diego-Carlsba...|\n",
      "|Cleveland-Elyria, OH|\n",
      "|Kingsport-Bristol...|\n",
      "|New Haven-Milford...|\n",
      "|Birmingham-Hoover...|\n",
      "|  Corpus Christi, TX|\n",
      "|         Dubuque, IA|\n",
      "|Las Vegas-Henders...|\n",
      "|Indianapolis-Carm...|\n",
      "|Seattle-Tacoma-Be...|\n",
      "|          Albany, OR|\n",
      "|   Winston-Salem, NC|\n",
      "|     Bakersfield, CA|\n",
      "|Los Angeles-Long ...|\n",
      "|Minneapolis-St. P...|\n",
      "|San Francisco-Oak...|\n",
      "|Phoenix-Mesa-Scot...|\n",
      "+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('location').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                page|\n",
      "+--------------------+\n",
      "|              Cancel|\n",
      "|    Submit Downgrade|\n",
      "|         Thumbs Down|\n",
      "|                Home|\n",
      "|           Downgrade|\n",
      "|         Roll Advert|\n",
      "|              Logout|\n",
      "|       Save Settings|\n",
      "|Cancellation Conf...|\n",
      "|               About|\n",
      "+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('page').distinct().show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exploratory Data Analysis\n",
    "\n",
    "### 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": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "52"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clean_data = df.select('artist','auth','firstName','gender','lastName','length','level','location','page','song','ts','userId')\n",
    "clean_data.where((col(\"page\") == \"Cancellation Confirmation\")).select(\"userId\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+---------+------+---------+------+-----+--------------------+--------------------+----+-------------+------+\n",
      "|artist|     auth|firstName|gender| lastName|length|level|            location|                page|song|           ts|userId|\n",
      "+------+---------+---------+------+---------+------+-----+--------------------+--------------------+----+-------------+------+\n",
      "|  null|Cancelled|   Adriel|     M|  Mendoza|  null| paid|  Kansas City, MO-KS|Cancellation Conf...|null|1538943990000|    18|\n",
      "|  null|Cancelled|    Diego|     M|    Mckee|  null| paid|Phoenix-Mesa-Scot...|Cancellation Conf...|null|1539033046000|    32|\n",
      "|  null|Cancelled|    Mason|     M|     Hart|  null| free|  Corpus Christi, TX|Cancellation Conf...|null|1539318918000|   125|\n",
      "|  null|Cancelled|Alexander|     M|   Garcia|  null| paid|Indianapolis-Carm...|Cancellation Conf...|null|1539375441000|   105|\n",
      "|  null|Cancelled|    Kayla|     F|  Johnson|  null| paid|Philadelphia-Camd...|Cancellation Conf...|null|1539465584000|    17|\n",
      "|  null|Cancelled|    Molly|     F| Harrison|  null| free|Virginia Beach-No...|Cancellation Conf...|null|1539588854000|   143|\n",
      "|  null|Cancelled|     Alex|     M|    Hogan|  null| paid|Denver-Aurora-Lak...|Cancellation Conf...|null|1539729037000|   101|\n",
      "|  null|Cancelled|    Davis|     M|     Wang|  null| paid|           Flint, MI|Cancellation Conf...|null|1539736161000|   129|\n",
      "|  null|Cancelled|  Nikolas|     M|    Olsen|  null| paid|Oxnard-Thousand O...|Cancellation Conf...|null|1539759749000|   121|\n",
      "|  null|Cancelled|    Ethan|     M|  Johnson|  null| paid|Lexington-Fayette...|Cancellation Conf...|null|1539761972000|    51|\n",
      "|  null|Cancelled|Christian|     M| Robinson|  null| paid|       Quincy, IL-MO|Cancellation Conf...|null|1540050556000|    87|\n",
      "|  null|Cancelled|    Molly|     F|Patterson|  null| paid|   Memphis, TN-MS-AR|Cancellation Conf...|null|1540062068000|   122|\n",
      "|  null|Cancelled|   Sophia|     F|    Perry|  null| paid|Los Angeles-Long ...|Cancellation Conf...|null|1540193374000|    12|\n",
      "|  null|Cancelled|    Erick|     M|   Brooks|  null| paid|           Selma, AL|Cancellation Conf...|null|1540223006000|    58|\n",
      "|  null|Cancelled|   Rachel|     F|   Bailey|  null| paid|Albany-Schenectad...|Cancellation Conf...|null|1540402387000|    73|\n",
      "|  null|Cancelled|  Jeffery|     M|  Wheeler|  null| paid|         Bozeman, MT|Cancellation Conf...|null|1540875543000|     3|\n",
      "|  null|Cancelled|   Sophia|     F|      Key|  null| paid|Los Angeles-Long ...|Cancellation Conf...|null|1541166424000|   106|\n",
      "|  null|Cancelled|    Piper|     F|  Nielsen|  null| paid|New York-Newark-J...|Cancellation Conf...|null|1541340091000|   103|\n",
      "|  null|Cancelled|   Teagan|     F|  Roberts|  null| paid|New Philadelphia-...|Cancellation Conf...|null|1541463632000|    28|\n",
      "|  null|Cancelled|    Alexi|     F|   Warren|  null| paid|Spokane-Spokane V...|Cancellation Conf...|null|1542051608000|    54|\n",
      "+------+---------+---------+------+---------+------+-----+--------------------+--------------------+----+-------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "clean_data.filter(clean_data.auth==\"Cancelled\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Filtering churned users\n",
    "user_churned = clean_data.filter(clean_data.auth==\"Cancelled\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+--------------------+---------+---------+------+---------+---------+-----+--------------------+-----------+--------------------+-------------+\n",
      "|userId|Churned|              artist|     auth|firstName|gender| lastName|   length|level|            location|       page|                song|           ts|\n",
      "+------+-------+--------------------+---------+---------+------+---------+---------+-----+--------------------+-----------+--------------------+-------------+\n",
      "|100010|  false|Sleeping With Sirens|Logged In| Darianna|     F|Carpenter|202.97098| free|Bridgeport-Stamfo...|   NextSong|Captain Tyin Knot...|1539003534000|\n",
      "|100010|  false|Francesca Battist...|Logged In| Darianna|     F|Carpenter|196.54485| free|Bridgeport-Stamfo...|   NextSong|Beautiful_ Beauti...|1539003736000|\n",
      "|100010|  false|              Brutha|Logged In| Darianna|     F|Carpenter|263.13098| free|Bridgeport-Stamfo...|   NextSong|          She's Gone|1539003932000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|  Thumbs Up|                null|1539003933000|\n",
      "|100010|  false|         Josh Ritter|Logged In| Darianna|     F|Carpenter|316.23791| free|Bridgeport-Stamfo...|   NextSong|      Folk Bloodbath|1539004195000|\n",
      "|100010|  false|               LMFAO|Logged In| Darianna|     F|Carpenter|183.74485| free|Bridgeport-Stamfo...|   NextSong|                 Yes|1539004511000|\n",
      "|100010|  false|         OneRepublic|Logged In| Darianna|     F|Carpenter|224.67873| free|Bridgeport-Stamfo...|   NextSong|             Secrets|1539004694000|\n",
      "|100010|  false|       Dwight Yoakam|Logged In| Darianna|     F|Carpenter| 239.3073| free|Bridgeport-Stamfo...|   NextSong|      You're The One|1539004918000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|Roll Advert|                null|1539004961000|\n",
      "|100010|  false|      The Chordettes|Logged In| Darianna|     F|Carpenter|142.41914| free|Bridgeport-Stamfo...|   NextSong|          Mr Sandman|1539005157000|\n",
      "|100010|  false|Coko featuring Ki...|Logged In| Darianna|     F|Carpenter| 249.3122| free|Bridgeport-Stamfo...|   NextSong|           I Get Joy|1539005299000|\n",
      "|100010|  false|            The Cure|Logged In| Darianna|     F|Carpenter| 52.27057| free|Bridgeport-Stamfo...|   NextSong|     The Final Sound|1539005548000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|Roll Advert|                null|1539005571000|\n",
      "|100010|  false|Kid Cudi Vs Crookers|Logged In| Darianna|     F|Carpenter|162.97751| free|Bridgeport-Stamfo...|   NextSong|        Day 'N' Nite|1539005600000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|  Thumbs Up|                null|1539005601000|\n",
      "|100010|  false|            Yeasayer|Logged In| Darianna|     F|Carpenter|323.44771| free|Bridgeport-Stamfo...|   NextSong|                2080|1539005762000|\n",
      "|100010|  false|             Ben Lee|Logged In| Darianna|     F|Carpenter|245.78567| free|Bridgeport-Stamfo...|   NextSong|        Ache For You|1539006085000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|Roll Advert|                null|1539006103000|\n",
      "|100010|  false|  ? & The Mysterians|Logged In| Darianna|     F|Carpenter|128.10404| free|Bridgeport-Stamfo...|   NextSong|    Just Like A Rose|1539006330000|\n",
      "|100010|  false|                null|Logged In| Darianna|     F|Carpenter|     null| free|Bridgeport-Stamfo...|Roll Advert|                null|1539006331000|\n",
      "+------+-------+--------------------+---------+---------+------+---------+---------+-----+--------------------+-----------+--------------------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Grouping by userId\n",
    "churn_df = clean_data.groupby('userId').agg(collect_list('auth').alias(\"auths\"))\n",
    "\n",
    "# Filtering churned user with lambda functions\n",
    "churned = udf(lambda x: 'Cancelled' in x)\n",
    "churn_df = churn_df.withColumn(\"Churned\", churned(churn_df.auths))\n",
    "\n",
    "# Dropping alias\n",
    "churn_df = churn_df.drop('auths')\n",
    "\n",
    "# Joining churned data with clean data\n",
    "label = churn_df.join(clean_data,'userId')\n",
    "\n",
    "label.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+\n",
      "|userId|Churned|\n",
      "+------+-------+\n",
      "|100010|  false|\n",
      "|200002|  false|\n",
      "|   125|   true|\n",
      "|   124|  false|\n",
      "|    51|   true|\n",
      "|     7|  false|\n",
      "|    15|  false|\n",
      "|    54|   true|\n",
      "|   155|  false|\n",
      "|   132|  false|\n",
      "|   154|  false|\n",
      "|100014|   true|\n",
      "|   101|   true|\n",
      "|    11|  false|\n",
      "|   138|  false|\n",
      "|300017|  false|\n",
      "|    29|   true|\n",
      "|    69|  false|\n",
      "|100021|   true|\n",
      "|   112|  false|\n",
      "+------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Churned user table\n",
    "churn_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "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",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "clean_data.printSchema()"
   ]
  },
  {
   "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."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+-------+\n",
      "|userId|th_up|th_down|\n",
      "+------+-----+-------+\n",
      "|100010|   17|      5|\n",
      "|200002|   21|      6|\n",
      "|   124|  171|     41|\n",
      "|    51|  100|     21|\n",
      "|     7|    7|      1|\n",
      "|    15|   81|     14|\n",
      "|    54|  163|     29|\n",
      "|   155|   58|      3|\n",
      "|100014|   17|      3|\n",
      "|   132|   96|     17|\n",
      "|   101|   86|     16|\n",
      "|    11|   40|      9|\n",
      "|   138|   95|     24|\n",
      "|300017|  303|     28|\n",
      "|100021|   11|      5|\n",
      "|    29|  154|     22|\n",
      "|    69|   72|      9|\n",
      "|   112|    9|      3|\n",
      "|    42|  166|     25|\n",
      "|    73|   14|      7|\n",
      "+------+-----+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Creating columns for like and dislike of a song\n",
    "th_up = label.where(label.page=='Thumbs Up').groupby(\"userId\").agg(count(col('page')).alias('th_up')).orderBy('userId')\n",
    "th_down = label.where(label.page=='Thumbs Down').groupby(\"userId\").agg(count(col('page')).alias('th_down')).orderBy('userId')\n",
    "like_dislike = th_up.join(th_down,'userId')\n",
    "like_dislike.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating column for number of songs played by user\n",
    "num_songs = label.where(col('song')!='null').groupby(\"userId\").agg(count(col('song')).alias('num_songs')).orderBy('userId')\n",
    "\n",
    "# Creating column for number of days a user has spent in the app\n",
    "duration = label.groupby('userId').agg(((max(col('ts')) - min(col('ts')))/86400000).alias(\"duration\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+\n",
      "|userId|num_songs|\n",
      "+------+---------+\n",
      "|    10|      673|\n",
      "|   100|     2682|\n",
      "|100001|      133|\n",
      "|100002|      195|\n",
      "|100003|       51|\n",
      "|100004|      942|\n",
      "|100005|      154|\n",
      "|100006|       26|\n",
      "|100007|      423|\n",
      "|100008|      772|\n",
      "|100009|      518|\n",
      "|100010|      275|\n",
      "|100011|       11|\n",
      "|100012|      476|\n",
      "|100013|     1131|\n",
      "|100014|      257|\n",
      "|100015|      800|\n",
      "|100016|      530|\n",
      "|100017|       52|\n",
      "|100018|     1002|\n",
      "+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "num_songs.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------------+\n",
      "|userId|           duration|\n",
      "+------+-------------------+\n",
      "|100010|  44.21780092592593|\n",
      "|200002| 45.496805555555554|\n",
      "|   125|0.02053240740740741|\n",
      "|   124| 59.996944444444445|\n",
      "|    51| 15.779398148148148|\n",
      "|     7| 50.784050925925925|\n",
      "|    15|  54.77318287037037|\n",
      "|    54|  42.79719907407407|\n",
      "|   155|  25.82783564814815|\n",
      "|100014| 41.244363425925926|\n",
      "|   132|  50.49740740740741|\n",
      "|   154| 24.986458333333335|\n",
      "|   101| 15.861481481481482|\n",
      "|    11| 53.241585648148146|\n",
      "|   138|  56.07674768518518|\n",
      "|300017|  59.11390046296296|\n",
      "|100021| 45.457256944444445|\n",
      "|    29|  43.32092592592593|\n",
      "|    69|  50.98648148148148|\n",
      "|   112|  56.87869212962963|\n",
      "+------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "duration.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+-----+-------+---------+------------------+\n",
      "|userId|Churned|th_up|th_down|num_songs|          duration|\n",
      "+------+-------+-----+-------+---------+------------------+\n",
      "|100010|  false|   17|      5|      275| 44.21780092592593|\n",
      "|200002|  false|   21|      6|      387|45.496805555555554|\n",
      "|   124|  false|  171|     41|     4079|59.996944444444445|\n",
      "|    51|   true|  100|     21|     2111|15.779398148148148|\n",
      "|     7|  false|    7|      1|      150|50.784050925925925|\n",
      "|    15|  false|   81|     14|     1914| 54.77318287037037|\n",
      "|    54|   true|  163|     29|     2841| 42.79719907407407|\n",
      "|   155|  false|   58|      3|      820| 25.82783564814815|\n",
      "|100014|   true|   17|      3|      257|41.244363425925926|\n",
      "|   132|  false|   96|     17|     1928| 50.49740740740741|\n",
      "|   101|   true|   86|     16|     1797|15.861481481481482|\n",
      "|    11|  false|   40|      9|      647|53.241585648148146|\n",
      "|   138|  false|   95|     24|     2070| 56.07674768518518|\n",
      "|300017|  false|  303|     28|     3632| 59.11390046296296|\n",
      "|100021|   true|   11|      5|      230|45.457256944444445|\n",
      "|    29|   true|  154|     22|     3028| 43.32092592592593|\n",
      "|    69|  false|   72|      9|     1125| 50.98648148148148|\n",
      "|   112|  false|    9|      3|      215| 56.87869212962963|\n",
      "|    42|  false|  166|     25|     3573| 60.08825231481482|\n",
      "|    73|   true|   14|      7|      377| 21.52954861111111|\n",
      "+------+-------+-----+-------+---------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Joining all the features\n",
    "final_features = churn_df.join(like_dislike,'userId')\n",
    "final_features = final_features.join(num_songs,'userId')\n",
    "final_features = final_features.join(duration,'userId')\n",
    "final_features.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To see if we got any null values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+-----+-------+--------+\n",
      "|userId|Churned|th_up|th_down|duration|\n",
      "+------+-------+-----+-------+--------+\n",
      "|     0|      0|    0|      0|       0|\n",
      "+------+-------+-----+-------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final_features.select([count(when(isnull(x), x)).alias(x) for x in [\"userId\", \"Churned\", \"th_up\", \"th_down\", \"duration\"]]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Adding more features\n",
    "up_song = udf(lambda Up, songs: float(Up)/float(songs), FloatType())\n",
    "down_song = udf(lambda Down, songs: float(Down)/float(songs), FloatType())\n",
    "song_hour = udf(lambda Songs, Days: float(Songs)/float((Days*24)), FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+-----+-------+---------+------------------+-----------+------------+-----------+\n",
      "|userId|Churned|th_up|th_down|num_songs|          duration|    up_song|   down_song|  song_hour|\n",
      "+------+-------+-----+-------+---------+------------------+-----------+------------+-----------+\n",
      "|100010|  false|   17|      5|      275| 44.21780092592593|0.061818182| 0.018181818| 0.25913393|\n",
      "|200002|  false|   21|      6|      387|45.496805555555554|0.054263566| 0.015503876| 0.35442048|\n",
      "|   124|  false|  171|     41|     4079|59.996944444444445| 0.04192204| 0.010051483|  2.8327832|\n",
      "|    51|   true|  100|     21|     2111|15.779398148148148|0.047370914| 0.009947892|  5.5742517|\n",
      "|     7|  false|    7|      1|      150|50.784050925925925|0.046666667| 0.006666667|0.123070136|\n",
      "|    15|  false|   81|     14|     1914| 54.77318287037037| 0.04231975|0.0073145246|  1.4560045|\n",
      "|    54|   true|  163|     29|     2841| 42.79719907407407|0.057374164| 0.010207674|   2.765952|\n",
      "|   155|  false|   58|      3|      820| 25.82783564814815| 0.07073171|0.0036585366|  1.3228621|\n",
      "|100014|   true|   17|      3|      257|41.244363425925926| 0.06614786|0.0116731515| 0.25963143|\n",
      "|   132|  false|   96|     17|     1928| 50.49740740740741|0.049792532| 0.008817428|  1.5908407|\n",
      "|   101|   true|   86|     16|     1797|15.861481481481482| 0.04785754| 0.008903729|  4.7205553|\n",
      "|    11|  false|   40|      9|      647|53.241585648148146|0.061823804| 0.013910355|  0.5063398|\n",
      "|   138|  false|   95|     24|     2070| 56.07674768518518| 0.04589372| 0.011594203|  1.5380707|\n",
      "|300017|  false|  303|     28|     3632| 59.11390046296296| 0.08342511|0.0077092513|  2.5600295|\n",
      "|100021|   true|   11|      5|      230|45.457256944444445|0.047826085|  0.02173913| 0.21082076|\n",
      "|    29|   true|  154|     22|     3028| 43.32092592592593|0.050858654|0.0072655217|  2.9123724|\n",
      "|    69|  false|   72|      9|     1125| 50.98648148148148|      0.064|       0.008| 0.91936135|\n",
      "|   112|  false|    9|      3|      215| 56.87869212962963|0.041860465| 0.013953488| 0.15749893|\n",
      "|    42|  false|  166|     25|     3573| 60.08825231481482| 0.04645956|0.0069969213|  2.4776058|\n",
      "|    73|   true|   14|      7|      377| 21.52954861111111|0.037135277| 0.018567638|  0.7296174|\n",
      "+------+-------+-----+-------+---------+------------------+-----------+------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Joining newly added features with final_features\n",
    "final_features = final_features.withColumn(\"up_song\", up_song(final_features.th_up, final_features.num_songs))\n",
    "final_features = final_features.withColumn(\"down_song\", down_song(final_features.th_down, final_features.num_songs))\n",
    "final_features = final_features.withColumn(\"song_hour\", song_hour(final_features.num_songs, final_features.duration))\n",
    "final_features.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Scale features\n",
    "- __VectorAssembler__ - VectorAssembler is a transformer that combines a given list of columns into a single vector column. It is useful for combining raw features and features generated by different feature transformers into a single feature vector, in order to train ML models like logistic regression and decision trees. VectorAssembler accepts the following input column types: all numeric types, boolean type, and vector type. In each row, the values of the input columns will be concatenated into a vector in the specified order.\n",
    "\n",
    "- __MinMaxScaler__ - transforms a dataset of Vector rows, rescaling each feature to a specific range (often [0, 1]). It takes parameters:\n",
    "\n",
    "\n",
    "    - min: 0.0 by default. Lower bound after transformation, shared by all features.\n",
    "    - max: 1.0 by default. Upper bound after transformation, shared by all features.\n",
    "    \n",
    "  MinMaxScaler computes summary statistics on a data set and produces a MinMaxScalerModel. The model can then transform each feature individually such that it is in the given range.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reference: Official Apache Spark Documentation (https://spark.apache.org/docs/latest/ml-features.html)\n",
    "assembler = VectorAssembler(\n",
    "    inputCols=[\"num_songs\", \"up_song\", \"down_song\", \"duration\", \"song_hour\"],\n",
    "    outputCol=\"vector_features\")\n",
    "final_features = assembler.transform(final_features)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Features scaled to range: [0.000000, 1.000000]\n",
      "+--------------------+--------------------+\n",
      "|     vector_features|      scaledFeatures|\n",
      "+--------------------+--------------------+\n",
      "|[275.0,0.06181818...|[0.03121865596790...|\n",
      "|[387.0,0.05426356...|[0.04526078234704...|\n",
      "|[4079.0,0.0419220...|[0.50814944834503...|\n",
      "|[2111.0,0.0473709...|[0.26140922768304...|\n",
      "|[150.0,0.04666666...|[0.01554663991975...|\n",
      "|[1914.0,0.0423197...|[0.23671013039117...|\n",
      "|[2841.0,0.0573741...|[0.35293380140421...|\n",
      "|[820.0,0.07073170...|[0.09954864593781...|\n",
      "|[257.0,0.06614785...|[0.02896188565697...|\n",
      "|[1928.0,0.0497925...|[0.23846539618856...|\n",
      "|[1797.0,0.0478575...|[0.22204112337011...|\n",
      "|[647.0,0.06182380...|[0.07785857572718...|\n",
      "|[2070.0,0.0458937...|[0.25626880641925...|\n",
      "|[3632.0,0.0834251...|[0.45210631895687...|\n",
      "|[230.0,0.04782608...|[0.02557673019057...|\n",
      "|[3028.0,0.0508586...|[0.37637913741223...|\n",
      "|[1125.0,0.0640000...|[0.13778836509528...|\n",
      "|[215.0,0.04186046...|[0.02369608826479...|\n",
      "|[3573.0,0.0464595...|[0.44470912738214...|\n",
      "|[377.0,0.03713527...|[0.04400702106318...|\n",
      "+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Reference: Official Apache Spark Documentation (https://spark.apache.org/docs/latest/ml-features.html)\n",
    "scaler = MinMaxScaler(inputCol=\"vector_features\", outputCol=\"scaledFeatures\")\n",
    "# Compute summary statistics and generate MinMaxScalerModel\n",
    "scalerModel = scaler.fit(final_features)\n",
    "\n",
    "# rescale each feature to range [min, max].\n",
    "final_features = scalerModel.transform(final_features)\n",
    "print(\"Features scaled to range: [%f, %f]\" % (scaler.getMin(), scaler.getMax()))\n",
    "final_features.select(\"vector_features\", \"scaledFeatures\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "int_conversion = udf(lambda a: 1 if a==\"true\" else 0, IntegerType())\n",
    "final_features = final_features.withColumn('label', int_conversion(final_features.Churned))"
   ]
  },
  {
   "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."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = final_features.randomSplit([0.8, 0.2], seed=42)\n",
    "train, validation = train.randomSplit([0.8, 0.2], seed=42)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Training\n",
    "- __Logistic Regression__ - Logistic regression is a classification algorithm used to assign observations to a discrete set of classes. Unlike linear regression which outputs continuous number values, logistic regression transforms its output using the logistic sigmoid function to return a probability value which can then be mapped to two or more discrete classes.\n",
    "\n",
    "Read more about [Area Under ROC here](http://gim.unmc.edu/dxtests/roc3.htm)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reference: Official Apache Spark Documentation (https://spark.apache.org/docs/2.1.1/ml-classification-regression.html)\n",
    "lr = LogisticRegression(featuresCol = 'vector_features', labelCol = 'label', maxIter=10)\n",
    "lrModel = lr.fit(train)\n",
    "trainingSummary = lrModel.summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "objectiveHistory:\n",
      "0.5218621245387386\n",
      "0.484219381160328\n",
      "0.4383255965606231\n",
      "0.40121483945570474\n",
      "0.39719928524762355\n",
      "0.38004849359549553\n",
      "0.3696234970965648\n",
      "0.3541276846797807\n",
      "0.3459847651158903\n",
      "0.3429506737673446\n",
      "0.33580669284130193\n",
      "+--------------------+-------------------+\n",
      "|                 FPR|                TPR|\n",
      "+--------------------+-------------------+\n",
      "|                 0.0|                0.0|\n",
      "|0.007874015748031496|                0.0|\n",
      "|0.007874015748031496|0.02857142857142857|\n",
      "|0.007874015748031496|0.05714285714285714|\n",
      "|0.007874015748031496|0.08571428571428572|\n",
      "|0.015748031496062992|0.08571428571428572|\n",
      "|0.015748031496062992|0.11428571428571428|\n",
      "|0.015748031496062992|0.14285714285714285|\n",
      "|0.015748031496062992|0.17142857142857143|\n",
      "|0.015748031496062992|                0.2|\n",
      "|0.015748031496062992|0.22857142857142856|\n",
      "|0.023622047244094488|0.22857142857142856|\n",
      "|0.023622047244094488| 0.2571428571428571|\n",
      "|0.023622047244094488| 0.2857142857142857|\n",
      "|0.023622047244094488| 0.3142857142857143|\n",
      "|0.023622047244094488|0.34285714285714286|\n",
      "|0.023622047244094488|0.37142857142857144|\n",
      "|0.023622047244094488|                0.4|\n",
      "|0.023622047244094488|0.42857142857142855|\n",
      "|0.031496062992125984|0.42857142857142855|\n",
      "+--------------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "areaUnderROC: 0.8929133858267712\n"
     ]
    }
   ],
   "source": [
    "# Reference: Official Apache Spark Documentation (https://spark.apache.org/docs/2.1.1/ml-classification-regression.html)\n",
    "objectiveHistory = trainingSummary.objectiveHistory\n",
    "print(\"objectiveHistory:\")\n",
    "for objective in objectiveHistory:\n",
    "    print(objective)\n",
    "\n",
    "# Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.\n",
    "trainingSummary.roc.show()\n",
    "print(\"areaUnderROC: \" + str(trainingSummary.areaUnderROC))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Best Threshold Value: 0.3818777091234896\n",
      "Max F-Measure: Row(max(F-Measure)=0.7222222222222223)\n",
      "+------------------+-------------------+\n",
      "|         threshold|          F-Measure|\n",
      "+------------------+-------------------+\n",
      "|0.8655645162024789|                0.0|\n",
      "|0.8603991487453458|0.05405405405405405|\n",
      "|0.8536839890357982|0.10526315789473684|\n",
      "|0.8491077815038794|0.15384615384615383|\n",
      "|0.8358627405384967|               0.15|\n",
      "|0.8322794248363419|0.19512195121951217|\n",
      "|0.8065574966643941|0.23809523809523808|\n",
      "|0.7787801655141297| 0.2790697674418604|\n",
      "|0.7503131673288331| 0.3181818181818182|\n",
      "|0.7305390950035154| 0.3555555555555555|\n",
      "| 0.685888225327694|0.34782608695652173|\n",
      "|0.6734502434126916| 0.3829787234042553|\n",
      "|0.6655922013855425|0.41666666666666663|\n",
      "|0.6432105555312088|0.44897959183673464|\n",
      "|0.6405194138258693|0.48000000000000004|\n",
      "| 0.636050039266382| 0.5098039215686275|\n",
      "|0.6290250299091301| 0.5384615384615384|\n",
      "|0.5981343652506612| 0.5660377358490566|\n",
      "|0.5927993574513184| 0.5555555555555555|\n",
      "|0.5854108624014036| 0.5818181818181818|\n",
      "+------------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Reference: Official Apache Spark Documentation (https://spark.apache.org/docs/2.1.1/ml-classification-regression.html)\n",
    "\n",
    "# Set the model threshold to maximize F-Measure\n",
    "fMeasure = trainingSummary.fMeasureByThreshold\n",
    "maxFMeasure = fMeasure.groupBy().max('F-Measure').select('max(F-Measure)').head()\n",
    "bestThreshold = fMeasure.where(fMeasure['F-Measure'] == maxFMeasure['max(F-Measure)']) \\\n",
    "    .select('threshold').head()['threshold']\n",
    "lr.setThreshold(bestThreshold)\n",
    "print(\"Best Threshold Value: {}\".format(bestThreshold))\n",
    "print(\"Max F-Measure: {}\".format(maxFMeasure))\n",
    "fMeasure.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------------+\n",
      "|             recall|         precision|\n",
      "+-------------------+------------------+\n",
      "|                0.0|               0.0|\n",
      "|                0.0|               0.0|\n",
      "|0.02857142857142857|               0.5|\n",
      "|0.05714285714285714|0.6666666666666666|\n",
      "|0.08571428571428572|              0.75|\n",
      "|0.08571428571428572|               0.6|\n",
      "|0.11428571428571428|0.6666666666666666|\n",
      "|0.14285714285714285|0.7142857142857143|\n",
      "|0.17142857142857143|              0.75|\n",
      "|                0.2|0.7777777777777778|\n",
      "|0.22857142857142856|               0.8|\n",
      "|0.22857142857142856|0.7272727272727273|\n",
      "| 0.2571428571428571|              0.75|\n",
      "| 0.2857142857142857|0.7692307692307693|\n",
      "| 0.3142857142857143|0.7857142857142857|\n",
      "|0.34285714285714286|               0.8|\n",
      "|0.37142857142857144|            0.8125|\n",
      "|                0.4|0.8235294117647058|\n",
      "|0.42857142857142855|0.8333333333333334|\n",
      "|0.42857142857142855|0.7894736842105263|\n",
      "+-------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Displaying Precision and Recall metrics\n",
    "trainingSummary.pr.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Prediction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+-----+-------+---------+-------------------+-----------+------------+-----------+--------------------+--------------------+-----+--------------------+--------------------+----------+\n",
      "|userId|Churned|th_up|th_down|num_songs|           duration|    up_song|   down_song|  song_hour|     vector_features|      scaledFeatures|label|       rawPrediction|         probability|prediction|\n",
      "+------+-------+-----+-------+---------+-------------------+-----------+------------+-----------+--------------------+--------------------+-----+--------------------+--------------------+----------+\n",
      "|     7|  false|    7|      1|      150| 50.784050925925925|0.046666667| 0.006666667|0.123070136|[150.0,0.04666666...|[0.01554663991975...|    0|[3.2603700087458,...|[0.96304396167816...|       0.0|\n",
      "|    54|   true|  163|     29|     2841|  42.79719907407407|0.057374164| 0.010207674|   2.765952|[2841.0,0.0573741...|[0.35293380140421...|    1|[0.82277671419076...|[0.69482544135708...|       0.0|\n",
      "|300018|  false|  132|     24|     1640|  50.64989583333333|  0.0804878| 0.014634146|  1.3491307|[1640.0,0.0804878...|[0.20235707121364...|    0|[1.62725108046284...|[0.83579271628285...|       0.0|\n",
      "|100006|   true|    2|      2|       26|0.06488425925925925| 0.07692308|  0.07692308|  16.696396|[26.0,0.076923079...|[0.0,0.6247859094...|    1|[-4.5828704155125...|[0.01012200014936...|       1.0|\n",
      "|300011|  false|  437|     41|     4619|  60.09164351851852| 0.09460922|0.0088763805|   3.202747|[4619.0,0.0946092...|[0.57585255767301...|    0|[1.28659203854218...|[0.78356979777810...|       0.0|\n",
      "|    88|  false|  122|     20|     2045|  56.55054398148148|  0.0596577| 0.009779952|  1.5067642|[2045.0,0.0596577...|[0.25313440320962...|    0|[2.47906631598346...|[0.92266119878278...|       0.0|\n",
      "|   123|  false|    5|      1|      150| 28.345613425925926|0.033333335| 0.006666667| 0.22049268|[150.0,0.03333333...|[0.01554663991975...|    0|[1.18639894876440...|[0.76609640054521...|       0.0|\n",
      "|    94|  false|    4|      1|      146|  50.56168981481481| 0.02739726| 0.006849315|0.120315075|[146.0,0.02739726...|[0.01504513540621...|    0|[3.34091939016643...|[0.96580621777658...|       0.0|\n",
      "|    80|  false|   11|      4|      367|  56.68069444444444|0.029972753| 0.010899182| 0.26978615|[367.0,0.02997275...|[0.04275325977933...|    0|[3.46717418147465...|[0.96973920462653...|       0.0|\n",
      "+------+-------+-----+-------+---------+-------------------+-----------+------------+-----------+--------------------+--------------------+-----+--------------------+--------------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prediction on validation set\n",
    "preds = lrModel.transform(validation)\n",
    "preds.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Matched Predicition: 8\n",
      "Total number of prediction: 9\n"
     ]
    }
   ],
   "source": [
    "print(\"Matched Predicition: {}\".format(preds.filter(preds.label == preds.prediction).count()))\n",
    "print(\"Total number of prediction: {}\".format(preds.count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy: 88.88888888888889\n"
     ]
    }
   ],
   "source": [
    "print(\"Accuracy: {}\".format((8/9)*100))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Area Under ROC 1.0\n"
     ]
    }
   ],
   "source": [
    "evaluator = BinaryClassificationEvaluator()\n",
    "print('Test Area Under ROC', evaluator.evaluate(preds))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tuning Hyperparameters\n",
    "An important task in Machine Learning is model selection, or using data to find the best model or parameters for a given task. This is also called tuning. Tuning may be done for individual Estimators such as `LogisticRegression`, or for entire `Pipelines` which include multiple algorithms, featurization, and other steps. Users can tune an entire `Pipeline` at once, rather than tuning each element in the `Pipeline` separately.\n",
    "\n",
    "- __CrossValidator__ - `CrossValidator` begins by splitting the dataset into a set of folds which are used as separate training and test datasets. E.g., with k=3 folds, `CrossValidator` will generate 3 (training, test) dataset pairs, each of which uses 2/3 of the data for training and 1/3 for testing. To evaluate a particular ParamMap, `CrossValidator` computes the average evaluation metric for the 3 Models produced by fitting the `Estimator` on the 3 different (training, test) dataset pairs. After identifying the best `ParamMap`, `CrossValidator` finally re-fits the `Estimator` using the best `ParamMap` and the entire dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr.setThreshold(bestThreshold)\n",
    "paramGrid = ParamGridBuilder().addGrid(lr.regParam, [0.1, 0.01]).build()\n",
    "cross_validator = CrossValidator(estimator=lr,\n",
    "                          estimatorParamMaps=paramGrid,\n",
    "                          evaluator=BinaryClassificationEvaluator(),\n",
    "                          numFolds=3)\n",
    "optimized_model = cross_validator.fit(train)\n",
    "tuning_preds = optimized_model.transform(validation)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Evaluation of tuned model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8\n",
      "9\n"
     ]
    }
   ],
   "source": [
    "print(tuning_preds.filter(tuning_preds.label == tuning_preds.prediction).count())\n",
    "print(tuning_preds.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy: 88.88888888888889\n"
     ]
    }
   ],
   "source": [
    "print(\"Accuracy: {}\".format((8/9)*100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area Under ROC After Tuning Hyperparameter 1.0\n"
     ]
    }
   ],
   "source": [
    "print('Area Under ROC After Tuning Hyperparameter', evaluator.evaluate(tuning_preds))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### On test set with optimized model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_preds = optimized_model.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy on test set (Optimized Model): 90.32258064516128\n"
     ]
    }
   ],
   "source": [
    "print(\"Accuracy on test set (Optimized Model): {}\".format((test_preds.filter(test_preds.label == test_preds.prediction).count()/test_preds.count())*100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area Under ROC on Test Set (Optimized Model): 0.8952380952380955\n"
     ]
    }
   ],
   "source": [
    "print('Area Under ROC on Test Set (Optimized Model): {}'.format(evaluator.evaluate(test_preds)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### On test set with normal model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "lrTest_preds = lrModel.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy on test set (Normal Model): 80.64516129032258\n"
     ]
    }
   ],
   "source": [
    "print(\"Accuracy on test set (Normal Model): {}\".format((lrTest_preds.filter(lrTest_preds.label == lrTest_preds.prediction).count()/lrTest_preds.count())*100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area Under ROC on Test Set (Normal Model): 0.8904761904761908\n"
     ]
    }
   ],
   "source": [
    "print('Area Under ROC on Test Set (Normal Model): {}'.format(evaluator.evaluate(lrTest_preds)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([0.0007, 5.9586, 81.6604, -0.0967, -0.1749])"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lrModel.coefficients"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Our Tuned Model performed well over normal one**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using coefficient of LogisticRegression model, we can derive the features which are contributing most to predict the churn of customer:\n",
    "\n",
    "- Number of Songs played\n",
    "- Average number of Thumbs Up\n",
    "- Average number of Thumbs Down "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Troubleshooting\n",
    "Having trouble in installing PySpark? Refer to this article: [Install Spark on Ubuntu (PySpark)](https://medium.com/@GalarnykMichael/install-spark-on-ubuntu-pyspark-231c45677de0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Credits\n",
    "\n",
    "__This article is written by [Rohit Swami](https://rohitswami.com). You can catch him up on [LinkedIn](https://www.linkedin.com/in/rowhitswami), [GitHub](https://github.com/rowhitswami), [Twitter](https://twitter.com/rowhitswami), and [Medium](https://medium.com/@rowhitswami/). Feel free to visit his personal website [www.rohitswami.com](https://rohitswami.com) for some cool projects.__\n",
    "\n",
    "> __This tutorial is intended to be a public resource. As such, if you see any glaring inaccuracies or if a critical topic is missing, please feel free to point it out or (preferably) submit a pull request to improve the tutorial. Also, we are always looking to improve the scope of this article. For anything feel free to mail us @ colearninglounge@gmail.com__\n",
    "\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}