Churn_Analysis_upd.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import required libraries\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "import pandas as pd\n",
    "spark = SparkSession.builder.appName('Churn').getOrCreate()\n",
    "\n",
    "\n",
    "# save the file location \n",
    "file_path = \"./churn.csv\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read the input file\n",
    "df = spark.read.csv(file_path, inferSchema = True, \n",
    "                    header = True, sep = \",\" ,\n",
    "                    nanValue = ' ', nullValue = ' ')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['churn',\n",
       " 'accountlength',\n",
       " 'internationalplan',\n",
       " 'voicemailplan',\n",
       " 'numbervmailmessages',\n",
       " 'totaldayminutes',\n",
       " 'totaldaycalls',\n",
       " 'totaldaycharge',\n",
       " 'totaleveminutes',\n",
       " 'totalevecalls',\n",
       " 'totalevecharge',\n",
       " 'totalnightminutes',\n",
       " 'totalnightcalls',\n",
       " 'totalnightcharge',\n",
       " 'totalintlminutes',\n",
       " 'totalintlcalls',\n",
       " 'totalintlcharge',\n",
       " 'numbercustomerservicecalls']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- churn: string (nullable = true)\n",
      " |-- accountlength: integer (nullable = true)\n",
      " |-- internationalplan: string (nullable = true)\n",
      " |-- voicemailplan: string (nullable = true)\n",
      " |-- numbervmailmessages: integer (nullable = true)\n",
      " |-- totaldayminutes: double (nullable = true)\n",
      " |-- totaldaycalls: integer (nullable = true)\n",
      " |-- totaldaycharge: double (nullable = true)\n",
      " |-- totaleveminutes: double (nullable = true)\n",
      " |-- totalevecalls: integer (nullable = true)\n",
      " |-- totalevecharge: double (nullable = true)\n",
      " |-- totalnightminutes: double (nullable = true)\n",
      " |-- totalnightcalls: integer (nullable = true)\n",
      " |-- totalnightcharge: double (nullable = true)\n",
      " |-- totalintlminutes: double (nullable = true)\n",
      " |-- totalintlcalls: integer (nullable = true)\n",
      " |-- totalintlcharge: double (nullable = true)\n",
      " |-- numbercustomerservicecalls: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Lets look at the schema from the file\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# In some situations, spark may not be able to infer the data schema as easily as above\n",
    "# especially in cases where read from json formats.\n",
    "# we can force it to enforce the schema which we need, while it reads the dataset.\n",
    "# Lets explore that a bit before we move on with our analysis\n",
    "\n",
    "#from pyspark.sql.types import StructField, StringType, IntegerType, StructType\n",
    "#data_schema = [StructField('age',IntegerType(), True),\n",
    "#               StructField('name', StringType(), True)]\n",
    "#final_struc = StructType(fields=data_schema)\n",
    "#\n",
    "# once we defined the final structure, we can just use it while reading the dataset\n",
    "#df = spark.read.json('data.json', schema=final_struc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------------+-----------------+-------------+-------------------+---------------+-------------+--------------+---------------+-------------+--------------+-----------------+---------------+----------------+----------------+--------------+---------------+--------------------------+\n",
      "|churn|accountlength|internationalplan|voicemailplan|numbervmailmessages|totaldayminutes|totaldaycalls|totaldaycharge|totaleveminutes|totalevecalls|totalevecharge|totalnightminutes|totalnightcalls|totalnightcharge|totalintlminutes|totalintlcalls|totalintlcharge|numbercustomerservicecalls|\n",
      "+-----+-------------+-----------------+-------------+-------------------+---------------+-------------+--------------+---------------+-------------+--------------+-----------------+---------------+----------------+----------------+--------------+---------------+--------------------------+\n",
      "|    0|            0|                0|            0|                  0|              0|            0|             0|              0|            0|             7|                0|              0|               6|               0|             0|              0|                         0|\n",
      "+-----+-------------+-----------------+-------------+-------------------+---------------+-------------+--------------+---------------+-------------+--------------+-----------------+---------------+----------------+----------------+--------------+---------------+--------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check for NUll Values\n",
    "from pyspark.sql.functions import isnan, when, count, col\n",
    "\n",
    "df.select([count(when(isnan(c) | col(c).isNull(),c)).alias(c) for c in df.columns]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We do see that some of the fields have null values so we will need to impute them \n",
    "# before feeding them into the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|churn|count|\n",
      "+-----+-----+\n",
      "|   No| 4293|\n",
      "|  Yes|  707|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the count of output variable in the dataset\n",
    "df.groupBy('churn').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+------------------+-----------------+------------------+\n",
      "|summary|    accountlength|    totaldaycharge| totalnightcharge|    totalevecharge|\n",
      "+-------+-----------------+------------------+-----------------+------------------+\n",
      "|  count|             5000|              5000|             4994|              4993|\n",
      "|   mean|         100.2586|30.649668000000023|9.017503003604375|17.053647105948343|\n",
      "| stddev|39.69455954726711| 9.162068691639355|2.273414183416672| 4.292381665441585|\n",
      "|    min|                1|               0.0|              0.0|               0.0|\n",
      "|    max|              243|             59.76|            17.77|             30.91|\n",
      "+-------+-----------------+------------------+-----------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the distribution of data for the AccountLength, totaldaycharge, totalevecharge, totalNightcharge\n",
    "df.select('accountlength', 'totaldaycharge', 'totalnightcharge', 'totalevecharge').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------+-----+-----------+\n",
      "|numbercustomerservicecalls|churn|churn_count|\n",
      "+--------------------------+-----+-----------+\n",
      "|                         0|  Yes|        121|\n",
      "|                         0|   No|        902|\n",
      "|                         1|  Yes|        190|\n",
      "|                         1|   No|       1596|\n",
      "|                         2|  Yes|        122|\n",
      "|                         2|   No|       1005|\n",
      "|                         3|  Yes|         73|\n",
      "|                         3|   No|        592|\n",
      "|                         4|  Yes|        111|\n",
      "|                         4|   No|        141|\n",
      "|                         5|  Yes|         58|\n",
      "|                         5|   No|         38|\n",
      "|                         6|  Yes|         22|\n",
      "|                         6|   No|         12|\n",
      "|                         7|  Yes|          7|\n",
      "|                         7|   No|          6|\n",
      "|                         8|  Yes|          1|\n",
      "|                         8|   No|          1|\n",
      "|                         9|  Yes|          2|\n",
      "+--------------------------+-----+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Lets create a sql view for the data frame so that we can run some queries interactively.\n",
    "# Get some insights on the effect of certain predictor variables on the dependent variable\n",
    "# First, lets look at churn count by TotalCustomerservicecalls\n",
    "df.createOrReplaceTempView(\"Churn_data\")\n",
    "query = \"select numbercustomerservicecalls, churn, count(*) as churn_count \\\n",
    "        from Churn_data \\\n",
    "        group by numbercustomerservicecalls, churn \\\n",
    "        order by numbercustomerservicecalls,churn desc, churn_count\"\n",
    "\n",
    "spark.sql(query).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-----+----------+\n",
      "|voicemailplan|churn|plan_count|\n",
      "+-------------+-----+----------+\n",
      "|          yes|   No|      1221|\n",
      "|           no|   No|      3072|\n",
      "|          yes|  Yes|       102|\n",
      "|           no|  Yes|       605|\n",
      "+-------------+-----+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Next lets look at Voicemail plan predictor varible with the churn variable\n",
    "\n",
    "query = \"select voicemailplan, churn, count(*) as plan_count \\\n",
    "         from Churn_data \\\n",
    "         group by voicemailplan, churn\"\n",
    "spark.sql(query).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---+---+\n",
      "|accountlength_churn| No|Yes|\n",
      "+-------------------+---+---+\n",
      "|                 69| 31|  5|\n",
      "|                138| 34|  5|\n",
      "|                101| 48|  7|\n",
      "|                 88| 40| 10|\n",
      "|                170|  8|  2|\n",
      "|                115| 39|  9|\n",
      "|                217|  3|  0|\n",
      "|                  5|  2|  0|\n",
      "|                120| 47|  5|\n",
      "|                202|  2|  0|\n",
      "|                 10|  3|  0|\n",
      "|                 56| 22|  2|\n",
      "|                142| 26|  1|\n",
      "|                153| 11|  1|\n",
      "|                174| 10|  2|\n",
      "|                185|  9|  1|\n",
      "|                 42| 24|  2|\n",
      "|                 24|  7|  3|\n",
      "|                 37| 13|  2|\n",
      "|                 25| 10|  2|\n",
      "+-------------------+---+---+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Let's Look at Account Length variable and see if it can show any insights in current form\n",
    "\n",
    "df.stat.crosstab('accountlength', 'churn').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+\n",
      "|summary|    accountlength|\n",
      "+-------+-----------------+\n",
      "|  count|             5000|\n",
      "|   mean|         100.2586|\n",
      "| stddev|39.69455954726711|\n",
      "|    min|                1|\n",
      "|    max|              243|\n",
      "+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Lets try a different view - Understand description statistics for this variable\n",
    "\n",
    "df.select('accountlength').describe().show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+----+---+\n",
      "|internationalplan_churn|  No|Yes|\n",
      "+-----------------------+----+---+\n",
      "|                     no|4019|508|\n",
      "|                    yes| 274|199|\n",
      "+-----------------------+----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Let us look closely at International plan Variable and understand how it is  \n",
    "# impacting churn\n",
    "\n",
    "df.stat.crosstab('internationalplan', 'churn').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This tells us that customers who have international plan tend to churn less."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets start building our Model - We will use the pipeline functionality in Spark ML to transform our dataset\n",
    "# As we have less data overall, I am choosing a train/test split of 0.8,0.2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Records for training: 4016\n",
      "Records for validation: 984\n"
     ]
    }
   ],
   "source": [
    "churn_df = df\n",
    "(train_data, test_data) = churn_df.randomSplit([0.8, 0.2], 24)\n",
    "\n",
    "print(\"Records for training: \" + str(train_data.count()))\n",
    "print(\"Records for validation: \" + str(test_data.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler\n",
    "\n",
    "catColumns = [\"internationalplan\", \"voicemailplan\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Let us set the stages for our pipeline and then transform the variables accordingly\n",
    "# The first stage for our Pipeline is the transformer which performs StringIndexing and categorical Encoding\n",
    "stages = []\n",
    "\n",
    "for catCol in catColumns:\n",
    "    \n",
    "    stringIndexer = StringIndexer(inputCol= catCol, outputCol=catCol + \"Index\")\n",
    "    encoder = OneHotEncoderEstimator(inputCols=[stringIndexer.getOutputCol()], outputCols = [catCol + \"catVec\"])\n",
    "    \n",
    "    stages += [stringIndexer, encoder]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The Second stage for our Transformer pipeline is the Imputer to impute missing values\n",
    "\n",
    "from pyspark.ml.feature import Imputer\n",
    "imputer = Imputer(inputCols = [\"totalevecharge\", \"totalnightcharge\"], \\\n",
    "                  outputCols=[\"out_totalevecharge\", \"out_totalnightcharge\"])\n",
    "\n",
    "stages += [imputer]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create bins for Account Length variable as using it as a numeric variable might not add much value\n",
    "\n",
    "from pyspark.ml.feature import QuantileDiscretizer\n",
    "\n",
    "acctlen_bin = QuantileDiscretizer(numBuckets=4, inputCol = \"accountlength\", outputCol=\"acctlen_bin\")\n",
    "\n",
    "stages += [acctlen_bin]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the label_Idx for the Output Column\n",
    "\n",
    "label_Idx = StringIndexer(inputCol= \"churn\", outputCol = \"label\")\n",
    "stages += [label_Idx]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Let us create a vector assembler which will convert the input values into array format which \n",
    "# can be fed to the model\n",
    "\n",
    "numericCols = [\"acctlen_bin\", \"numbervmailmessages\", \"totaldayminutes\",\"totaldaycalls\", \\\n",
    "               \"totaldaycharge\", \"totaleveminutes\", \"totalevecalls\", \\\n",
    "               \"out_totalevecharge\", \"totalnightminutes\", \"totalnightcalls\", \\\n",
    "               \"out_totalnightcharge\", \"totalintlminutes\", \"totalintlcalls\", \\\n",
    "               \"totalintlcharge\", \"numbercustomerservicecalls\"]\n",
    "inputFeatures = [c + \"catVec\" for c in catColumns] + numericCols\n",
    "assembler = VectorAssembler(inputCols=inputFeatures, outputCol=\"features\")\n",
    "stages += [assembler]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[StringIndexer_382ffd84ebd4,\n",
       " OneHotEncoderEstimator_eb0073c4b65b,\n",
       " StringIndexer_adf0adb2e342,\n",
       " OneHotEncoderEstimator_45c993d4a763,\n",
       " Imputer_b1b0da79b1ab,\n",
       " QuantileDiscretizer_e221cc75a074,\n",
       " StringIndexer_d9ef30bc785c,\n",
       " VectorAssembler_1cb58fdaf39c]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create our stages into a pipeline\n",
    "pipeline = Pipeline().setStages(stages)\n",
    "pipelineModel = pipeline.fit(train_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run our data through the pipeline\n",
    "\n",
    "trainfinalDF = pipelineModel.transform(train_data)\n",
    "testfinalDF = pipelineModel.transform(test_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets fit a Random forest Algorithm to get insights on feature importances\n",
    "\n",
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "\n",
    "rf = RandomForestClassifier(labelCol=\"label\", featuresCol=\"features\")\n",
    "\n",
    "rf_model = rf.fit(trainfinalDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('totaldaycharge', 0.21485353351969486),\n",
       " ('totaldayminutes', 0.16679368310247106),\n",
       " ('numbercustomerservicecalls', 0.14772046100045796),\n",
       " ('internationalplancatVec', 0.09774993907102503),\n",
       " ('totaleveminutes', 0.06741407409952797),\n",
       " ('totalintlcalls', 0.052604088424604845),\n",
       " ('totalintlminutes', 0.05076341770100635),\n",
       " ('voicemailplancatVec', 0.049605674517355855),\n",
       " ('totalintlcharge', 0.04128566467198688),\n",
       " ('out_totalevecharge', 0.03524340854747537),\n",
       " ('numbervmailmessages', 0.022639067948644623),\n",
       " ('totalnightminutes', 0.019059749971999992),\n",
       " ('out_totalnightcharge', 0.015985548654851377),\n",
       " ('totalnightcalls', 0.0055304158138380945),\n",
       " ('totalevecalls', 0.0052188884977352355),\n",
       " ('totaldaycalls', 0.004925527496499302),\n",
       " ('acctlen_bin', 0.0026068569608253263)]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Look at feature importance value for the features\n",
    "\n",
    "feature_dict = {}\n",
    "for feature, imp in zip(inputFeatures, rf_model.featureImportances):\n",
    "    feature_dict[feature] = feature_dict.get(feature,0) + imp\n",
    "    \n",
    "feature_list = sorted(feature_dict.items(), key = lambda x:x[1], reverse = True)\n",
    "\n",
    "feature_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Let us rebuild the pipeline only considering the high significant features above\n",
    "\n",
    "stages = []\n",
    "\n",
    "for catCol in catColumns:\n",
    "    \n",
    "    stringIndexer = StringIndexer(inputCol= catCol, outputCol=catCol + \"Index\")\n",
    "    encoder = OneHotEncoderEstimator(inputCols=[stringIndexer.getOutputCol()], outputCols = [catCol + \"catVec\"])\n",
    "    \n",
    "    stages += [stringIndexer, encoder]\n",
    "\n",
    "imputer = Imputer(inputCols = [\"totalevecharge\", \"totalnightcharge\"], \\\n",
    "                  outputCols=[\"out_totalevecharge\", \"out_totalnightcharge\"])\n",
    "\n",
    "stages += [imputer]\n",
    "\n",
    "label_Idx = StringIndexer(inputCol= \"churn\", outputCol = \"label\")\n",
    "stages += [label_Idx]\n",
    "\n",
    "numericCols = [\"numbercustomerservicecalls\", \"totaldaycharge\", \"totaldayminutes\",\\\n",
    "               \"totaleveminutes\", \"out_totalevecharge\", \"totalintlcalls\"]\n",
    "inputFeatures = [c + \"catVec\" for c in catColumns] + numericCols\n",
    "assembler = VectorAssembler(inputCols=inputFeatures, outputCol=\"features\")\n",
    "stages += [assembler]\n",
    "\n",
    "# Create our stages into a pipeline\n",
    "pipeline = Pipeline().setStages(stages)\n",
    "pipelineModel = pipeline.fit(train_data)\n",
    "\n",
    "# Run our data through the pipeline\n",
    "\n",
    "trainfinalDF = pipelineModel.transform(train_data)\n",
    "testfinalDF = pipelineModel.transform(test_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets fit the Logistic Regression Model \n",
    "\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "logistic = LogisticRegression(labelCol=\"label\", featuresCol=\"features\", maxIter = 150)\n",
    "\n",
    "#Train the above model with our training data\n",
    "logistic_model = logistic.fit(trainfinalDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area under PR =  0.33972125435540074\n",
      "Area under ROC =  0.5589878131347326\n"
     ]
    }
   ],
   "source": [
    "# Lets predict with the test dataset and look at model metrics\n",
    "\n",
    "from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
    "\n",
    "predictions = logistic_model.transform(testfinalDF) # Outputs a df object\n",
    "\n",
    "results = predictions.select(['prediction', 'label'])\n",
    "res = results.collect()\n",
    "results_list = [(float(i[0]), float(i[1])) for i in res]\n",
    "preds_labels = spark.sparkContext.parallelize(results_list) # Convert to RDD to work with BCM\n",
    "\n",
    "metrics = BinaryClassificationMetrics(preds_labels)\n",
    "\n",
    "print(\"Area under PR = \", metrics.areaUnderPR)\n",
    "print(\"Area under ROC = \", metrics.areaUnderROC) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Correct: 843\n",
      "Wrong: 141\n",
      "tp: 20\n",
      "fp: 21\n",
      "fn: 120\n",
      "tn: 823\n",
      "Accuracy: 0.8567073170731707\n",
      "Precision: 0.4878048780487805\n",
      "Recall: 0.14285714285714285\n"
     ]
    }
   ],
   "source": [
    "# Let us look at Precision and Recall Metrics\n",
    "\n",
    "count=predictions.count()\n",
    "correct = results.filter(results['prediction'] == results['label']).count()\n",
    "wrong = results.filter(results['prediction'] != results['label']).count()\n",
    "tp = results.filter(results['prediction'] == 1.0).filter(results['prediction'] == results['label']).count()\n",
    "fp = results.filter(results['prediction'] == 1.0).filter(results['prediction'] != results['label']).count()\n",
    "fn = results.filter(results['prediction'] == 0.0).filter(results['prediction'] != results['label']).count()\n",
    "tn = results.filter(results['prediction'] == 0.0).filter(results['prediction'] == results['label']).count()\n",
    "\n",
    "accuracy = (tp+tn)/count\n",
    "\n",
    "precision = tp/(tp+fp)\n",
    "\n",
    "recall = tp/(tp+fn)\n",
    "\n",
    "print(\"Correct: %s\\nWrong: %s\\ntp: %s\\nfp: %s\\nfn: %s\\ntn: %s\\nAccuracy: %s\\nPrecision: %s\\nRecall: %s\"\n",
    "      % (correct, wrong, tp, fp, fn, tn, accuracy, precision, recall))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "# Create ParamGrid for Cross Validation\n",
    "paramGrid = (ParamGridBuilder()\n",
    "             .addGrid(logistic.regParam, [0.01, 0.1])\n",
    "             .build())\n",
    "\n",
    "evaluatorLR = BinaryClassificationEvaluator(rawPredictionCol=\"prediction\")\n",
    "\n",
    "cv = CrossValidator(estimator=logistic, estimatorParamMaps=paramGrid, evaluator=evaluatorLR, numFolds=3)\n",
    "\n",
    "cvModel = cv.fit(trainfinalDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.5553994583615437"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions = cvModel.bestModel.transform(testfinalDF)\n",
    "evaluatorLR.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Correct: 847\n",
      "Wrong: 137\n",
      "tp: 18\n",
      "fp: 15\n",
      "fn: 122\n",
      "tn: 829\n",
      "Accuracy: 0.8607723577235772\n",
      "Precision: 0.5454545454545454\n",
      "Recall: 0.12857142857142856\n"
     ]
    }
   ],
   "source": [
    "# Calculate the Accuracy, Precision and recall metrics again\n",
    "\n",
    "results = predictions.select(['prediction', 'label'])\n",
    "\n",
    "count=predictions.count()\n",
    "correct = results.filter(results.prediction == results.label).count()\n",
    "wrong = results.filter(results.prediction != results.label).count()\n",
    "tp = results.filter(results.prediction == 1.0).filter(results.prediction == results.label).count()\n",
    "fp = results.filter(results.prediction == 1.0).filter(results.prediction != results.label).count()\n",
    "fn = results.filter(results.prediction == 0.0).filter(results.prediction != results.label).count()\n",
    "tn = results.filter(results.prediction == 0.0).filter(results.prediction == results.label).count()\n",
    "\n",
    "accuracy = (tp+tn)/count\n",
    "\n",
    "precision = tp/(tp+fp)\n",
    "\n",
    "recall = tp/(tp+fn)\n",
    "\n",
    "print(\"Correct: %s\\nWrong: %s\\ntp: %s\\nfp: %s\\nfn: %s\\ntn: %s\\nAccuracy: %s\\nPrecision: %s\\nRecall: %s\"\n",
    "      % (correct, wrong, tp, fp, fn, tn, accuracy, precision, recall))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:spark2.7] *",
   "language": "python",
   "name": "conda-env-spark2.7-py"
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}