Credit Card Default.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Python for Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import sys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Setup the folder paths\n",
    "# Where you downloaded the resource bundle\n",
    "os.chdir(\"/Users/jlyang/Documents/Intern&Job/Spark_Python_Do_Big_Data_Analytics\")\n",
    "# Where you installed spark\n",
    "os.environ['SPARK_HOME'] = '/Users/jlyang/Spark/spark-2.1.0-bin-hadoop2.7'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create a variable for our root path\n",
    "SPARK_HOME = os.environ['SPARK_HOME']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Add the following paths to the system path\n",
    "sys.path.insert(0, os.path.join(SPARK_HOME,\"python\"))\n",
    "sys.path.insert(0, os.path.join(SPARK_HOME,\"python\",\"lib\"))\n",
    "sys.path.insert(0, os.path.join(SPARK_HOME,\"python\",\"lib\",\"pyspark.zip\"))\n",
    "sys.path.insert(0, os.path.join(SPARK_HOME,\"python\",\"lib\",\"py4j-0.10.4-src.zip\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Initialize SparkSession\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create a Spark Session\n",
    "SpSession = SparkSession.builder.master(\"local[2]\").appName(\"jlyang_spark\") \\\n",
    "            .config(\"spark.executor.memory\", \"1g\") \\\n",
    "            .config(\"spark.cores.max\",\"2\") \\\n",
    "            .config(\"spark.sql.warehouse.dir\", \"/Users/jlyang/Spark/spark-warehouse\")\\\n",
    "            .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Get the Spark Context from Spark Session    \n",
    "SpContext = SpSession.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data from the data file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'CUSTID,LIMIT_BAL,SEX,EDUCATION,MARRIAGE,AGE,PAY_1,PAY_2,PAY_3,PAY_4,PAY_5,PAY_6,BILL_AMT1,BILL_AMT2,BILL_AMT3,BILL_AMT4,BILL_AMT5,BILL_AMT6,PAY_AMT1,PAY_AMT2,PAY_AMT3,PAY_AMT4,PAY_AMT5,PAY_AMT6,DEFAULTED',\n",
       " u'530,20000,2,2,2,21,-1,-1,2,2,-2,-2,0,0,0,0,0,0,0,0,0,0,162000,0,0',\n",
       " u'38,60000,2,2,2,22,0,0,0,0,-2,-2,0,0,0,0,0,0,0,0,0,0,0,1576,0',\n",
       " u'43,10000,1,2,2,22,0,0,0,0,-2,-2,0,0,0,0,0,0,0,0,0,0,0,1500,0',\n",
       " u'47,20000,2,1,2,22,0,0,2,-1,0,-1,1131,291,582,291,0,291,291,582,0,0,130291,651,0']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Load the file into a RDD\n",
    "ccRaw = SpContext.textFile('credit-card-default-1000.csv')\n",
    "ccRaw.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1002"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Remove header row\n",
    "dataLines = ccRaw.filter(lambda x: 'CUSTID' not in x)\n",
    "dataLines.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Prepare and augment the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Cleanup data. Remove the last two lines which are not 'CSV'\n",
    "cleanedLines = dataLines.filter(lambda x: x.find('aaa') == -1)\n",
    "cleanedLines.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Convert into SQL Dataframe. In the process perform a few cleanups and changes required for future work\n",
    "def convertToRow(instr):\n",
    "    attList = instr.split(',')\n",
    "    \n",
    "    # PR#06 Round of age to range of 10s\n",
    "    ageRound = round(float(attList[5]) / 10) * 10\n",
    "    \n",
    "    # Normalize sex to only 1 and 2\n",
    "    sex = float(attList[2].replace('M', '1').replace('F', '2'))\n",
    "    \n",
    "    # Find average billed amount\n",
    "    avgBillAmt = np.array(attList[12:18], dtype = 'float').mean().item()\n",
    "    \n",
    "    # Find average pay amount\n",
    "    avgPayAmt = np.array(attList[18:24], dtype = 'float').mean().item()\n",
    "    \n",
    "    # Find average pay duration, required for PR#04\n",
    "    # Make sure numbers are rounded and negative values are eliminated\n",
    "    payDuration = np.array([x if float(x) > 0 else 0 for x in attList[6:12]], dtype = 'float')\n",
    "    avgPayDuration = round(payDuration.mean().item())\n",
    "    \n",
    "    # Average percentage paid. Add this as an additional field to see if this field has any predictive capabilities\n",
    "    perPay = round((avgPayAmt / (avgBillAmt + 1) * 100) / 25) * 25\n",
    "    \n",
    "    values = Row(CUSTID = attList[0], LIMIT_BAL = float(attList[1]), SEX = sex, EDUCATION = float(attList[3]), \\\n",
    "                 MARRIAGE = float(attList[4]), AGE = ageRound, AVG_PAY_DUR = avgPayDuration, \\\n",
    "                 AVG_BILL_AMT = avgBillAmt, AVG_PAY_AMT = avgPayAmt, PER_PAID = perPay, DEFAULTED = float(attList[24]))\n",
    "    \n",
    "    return values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(AGE=20.0, AVG_BILL_AMT=0.0, AVG_PAY_AMT=27000.0, AVG_PAY_DUR=1.0, CUSTID=u'530', DEFAULTED=0.0, EDUCATION=2.0, LIMIT_BAL=20000.0, MARRIAGE=2.0, PER_PAID=2700000.0, SEX=2.0),\n",
       " Row(AGE=20.0, AVG_BILL_AMT=0.0, AVG_PAY_AMT=262.6666666666667, AVG_PAY_DUR=0.0, CUSTID=u'38', DEFAULTED=0.0, EDUCATION=2.0, LIMIT_BAL=60000.0, MARRIAGE=2.0, PER_PAID=26275.0, SEX=2.0),\n",
       " Row(AGE=20.0, AVG_BILL_AMT=0.0, AVG_PAY_AMT=250.0, AVG_PAY_DUR=0.0, CUSTID=u'43', DEFAULTED=0.0, EDUCATION=2.0, LIMIT_BAL=10000.0, MARRIAGE=2.0, PER_PAID=25000.0, SEX=1.0),\n",
       " Row(AGE=20.0, AVG_BILL_AMT=431.0, AVG_PAY_AMT=21969.166666666668, AVG_PAY_DUR=0.0, CUSTID=u'47', DEFAULTED=0.0, EDUCATION=1.0, LIMIT_BAL=20000.0, MARRIAGE=2.0, PER_PAID=5075.0, SEX=2.0),\n",
       " Row(AGE=20.0, AVG_BILL_AMT=3349.5, AVG_PAY_AMT=28651.5, AVG_PAY_DUR=0.0, CUSTID=u'70', DEFAULTED=0.0, EDUCATION=4.0, LIMIT_BAL=20000.0, MARRIAGE=2.0, PER_PAID=850.0, SEX=1.0)]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Cleanedup RDD\n",
    "ccRows = cleanedLines.map(convertToRow)\n",
    "ccRows.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+------------------+-----------+------+---------+---------+---------+--------+---------+---+\n",
      "| AGE|      AVG_BILL_AMT|       AVG_PAY_AMT|AVG_PAY_DUR|CUSTID|DEFAULTED|EDUCATION|LIMIT_BAL|MARRIAGE| PER_PAID|SEX|\n",
      "+----+------------------+------------------+-----------+------+---------+---------+---------+--------+---------+---+\n",
      "|20.0|               0.0|           27000.0|        1.0|   530|      0.0|      2.0|  20000.0|     2.0|2700000.0|2.0|\n",
      "|20.0|               0.0| 262.6666666666667|        0.0|    38|      0.0|      2.0|  60000.0|     2.0|  26275.0|2.0|\n",
      "|20.0|               0.0|             250.0|        0.0|    43|      0.0|      2.0|  10000.0|     2.0|  25000.0|1.0|\n",
      "|20.0|             431.0|21969.166666666668|        0.0|    47|      0.0|      1.0|  20000.0|     2.0|   5075.0|2.0|\n",
      "|20.0|            3349.5|           28651.5|        0.0|    70|      0.0|      4.0|  20000.0|     2.0|    850.0|1.0|\n",
      "|20.0|1025.3333333333333|            7358.0|        0.0|    79|      0.0|      2.0|  30000.0|     2.0|    725.0|2.0|\n",
      "|20.0|117.83333333333333|             829.5|        0.0|    99|      0.0|      3.0|  50000.0|     1.0|    700.0|2.0|\n",
      "|20.0| 473.3333333333333|3328.3333333333335|        0.0|   104|      0.0|      3.0|  50000.0|     2.0|    700.0|2.0|\n",
      "|20.0|61.333333333333336| 359.8333333333333|        0.0|   135|      0.0|      2.0|  30000.0|     2.0|    575.0|2.0|\n",
      "|20.0|1316.8333333333333|            6896.5|        0.0|   170|      0.0|      2.0|  50000.0|     2.0|    525.0|2.0|\n",
      "+----+------------------+------------------+-----------+------+---------+---------+---------+--------+---------+---+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create a dataframe\n",
    "ccDf = SpSession.createDataFrame(ccRows)\n",
    "ccDf.cache()\n",
    "ccDf.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- AGE: double (nullable = true)\n",
      " |-- AVG_BILL_AMT: double (nullable = true)\n",
      " |-- AVG_PAY_AMT: double (nullable = true)\n",
      " |-- AVG_PAY_DUR: double (nullable = true)\n",
      " |-- CUSTID: string (nullable = true)\n",
      " |-- DEFAULTED: double (nullable = true)\n",
      " |-- EDUCATION: double (nullable = true)\n",
      " |-- LIMIT_BAL: double (nullable = true)\n",
      " |-- MARRIAGE: double (nullable = true)\n",
      " |-- PER_PAID: double (nullable = true)\n",
      " |-- SEX: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ccDf.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "|SEX|SEX_NAME|\n",
      "+---+--------+\n",
      "|1.0|    Male|\n",
      "|2.0|  Female|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Add SEX_NAME to the data using SQL join. Required for PR#02\n",
    "genderDf = SpSession.createDataFrame(pd.DataFrame({'SEX': [1.0, 2.0], 'SEX_NAME': ['Male', 'Female']}))\n",
    "genderDf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------------+-----------------+-----------+------+---------+---------+---------+--------+--------+---+--------+\n",
      "| AGE|       AVG_BILL_AMT|      AVG_PAY_AMT|AVG_PAY_DUR|CUSTID|DEFAULTED|EDUCATION|LIMIT_BAL|MARRIAGE|PER_PAID|SEX|SEX_NAME|\n",
      "+----+-------------------+-----------------+-----------+------+---------+---------+---------+--------+--------+---+--------+\n",
      "|70.0| -65.66666666666667|416.6666666666667|        0.0|   388|      1.0|      3.0|  80000.0|     1.0|  -650.0|1.0|    Male|\n",
      "|60.0|-56043.166666666664|          57956.5|        0.0|   103|      1.0|      1.0| 480000.0|     1.0|  -100.0|1.0|    Male|\n",
      "|60.0|                0.0|              0.0|        0.0|   932|      1.0|      1.0| 320000.0|     1.0|     0.0|1.0|    Male|\n",
      "|60.0|                0.0|              0.0|        0.0|   948|      1.0|      2.0|  50000.0|     1.0|     0.0|1.0|    Male|\n",
      "|60.0| 25828.333333333332|              0.0|        0.0|   602|      1.0|      3.0|  30000.0|     1.0|     0.0|1.0|    Male|\n",
      "+----+-------------------+-----------------+-----------+------+---------+---------+---------+--------+--------+---+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ccDf1 = ccDf.join(genderDf, ccDf.SEX == genderDf.SEX).drop(genderDf.SEX)\n",
    "ccDf1.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----------+\n",
      "|EDUCATION|     ED_STR|\n",
      "+---------+-----------+\n",
      "|      1.0|   Graduate|\n",
      "|      2.0| University|\n",
      "|      3.0|High School|\n",
      "|      4.0|     Others|\n",
      "+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Add ED_STR to the data using SQL join. Required for PR#03\n",
    "eduDf = SpSession.createDataFrame(pd.DataFrame({'EDUCATION': [1.0, 2.0, 3.0, 4.0], \n",
    "                                                'ED_STR': ['Graduate', 'University', 'High School', 'Others']}))\n",
    "eduDf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+\n",
      "| AGE|       AVG_BILL_AMT|AVG_PAY_AMT|AVG_PAY_DUR|CUSTID|DEFAULTED|EDUCATION|LIMIT_BAL|MARRIAGE|PER_PAID|SEX|SEX_NAME|  ED_STR|\n",
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+\n",
      "|60.0|-56043.166666666664|    57956.5|        0.0|   103|      1.0|      1.0| 480000.0|     1.0|  -100.0|1.0|    Male|Graduate|\n",
      "|60.0|                0.0|        0.0|        0.0|   932|      1.0|      1.0| 320000.0|     1.0|     0.0|1.0|    Male|Graduate|\n",
      "|60.0|                0.0|        0.0|        0.0|   466|      1.0|      1.0| 230000.0|     1.0|     0.0|1.0|    Male|Graduate|\n",
      "|60.0|            19763.0|        0.0|        1.0|    35|      1.0|      1.0| 500000.0|     1.0|     0.0|1.0|    Male|Graduate|\n",
      "|60.0|              -87.5|        0.0|        0.0|    66|      1.0|      1.0| 200000.0|     1.0|    -0.0|1.0|    Male|Graduate|\n",
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ccDf2 = ccDf1.join(eduDf, ccDf1.EDUCATION == eduDf.EDUCATION).drop(eduDf.EDUCATION)\n",
    "ccDf2.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|MARRIAGE|MARR_DESC|\n",
      "+--------+---------+\n",
      "|     1.0|   Single|\n",
      "|     2.0|  Married|\n",
      "|     3.0|   Others|\n",
      "+--------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Add MARR_DESC to the data with SQL join. Required for PR#03\n",
    "marrDf = SpSession.createDataFrame(pd.DataFrame({'MARRIAGE': [1.0, 2.0, 3.0], \n",
    "                                                 'MARR_DESC': ['Single', 'Married', 'Others']}))\n",
    "marrDf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+---------+\n",
      "| AGE|       AVG_BILL_AMT|AVG_PAY_AMT|AVG_PAY_DUR|CUSTID|DEFAULTED|EDUCATION|LIMIT_BAL|MARRIAGE|PER_PAID|SEX|SEX_NAME|  ED_STR|MARR_DESC|\n",
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+---------+\n",
      "|60.0|-56043.166666666664|    57956.5|        0.0|   103|      1.0|      1.0| 480000.0|     1.0|  -100.0|1.0|    Male|Graduate|   Single|\n",
      "|60.0|                0.0|        0.0|        0.0|   932|      1.0|      1.0| 320000.0|     1.0|     0.0|1.0|    Male|Graduate|   Single|\n",
      "|60.0|                0.0|        0.0|        0.0|   466|      1.0|      1.0| 230000.0|     1.0|     0.0|1.0|    Male|Graduate|   Single|\n",
      "|60.0|            19763.0|        0.0|        1.0|    35|      1.0|      1.0| 500000.0|     1.0|     0.0|1.0|    Male|Graduate|   Single|\n",
      "|60.0|              -87.5|        0.0|        0.0|    66|      1.0|      1.0| 200000.0|     1.0|    -0.0|1.0|    Male|Graduate|   Single|\n",
      "+----+-------------------+-----------+-----------+------+---------+---------+---------+--------+--------+---+--------+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ccFinalDf = ccDf2.join(marrDf, ccDf2.MARRIAGE == marrDf.MARRIAGE).drop(marrDf.MARRIAGE)\n",
    "ccFinalDf.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Do analysis as required by the problem statement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create a temp view\n",
    "ccFinalDf.createOrReplaceTempView('CCDATA')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+--------+-----------+\n",
      "|SEX_NAME|Total|Defaults|PER_DEFAULT|\n",
      "+--------+-----+--------+-----------+\n",
      "|  Female|  591|   218.0|       37.0|\n",
      "|    Male|  409|   185.0|       45.0|\n",
      "+--------+-----+--------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# PR#02 solution\n",
    "SpSession.sql('SELECT SEX_NAME, count(*) AS Total, SUM(DEFAULTED) AS Defaults, ' + \\\n",
    "             'ROUND(SUM(DEFAULTED) * 100 / count(*)) AS PER_DEFAULT FROM CCDATA GROUP BY SEX_NAME').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----------+-----+--------+-----------+\n",
      "|MARR_DESC|     ED_STR|Total|Defaults|PER_DEFAULT|\n",
      "+---------+-----------+-----+--------+-----------+\n",
      "|  Married|   Graduate|  268|    69.0|       26.0|\n",
      "|  Married|High School|   55|    24.0|       44.0|\n",
      "|  Married|     Others|    4|     2.0|       50.0|\n",
      "|  Married| University|  243|    65.0|       27.0|\n",
      "|   Others|   Graduate|    4|     4.0|      100.0|\n",
      "|   Others|High School|    8|     6.0|       75.0|\n",
      "|   Others| University|    7|     3.0|       43.0|\n",
      "|   Single|   Graduate|  123|    71.0|       58.0|\n",
      "|   Single|High School|   87|    52.0|       60.0|\n",
      "|   Single|     Others|    3|     2.0|       67.0|\n",
      "|   Single| University|  198|   105.0|       53.0|\n",
      "+---------+-----------+-----+--------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# PR#03 solution\n",
    "SpSession.sql('SELECT MARR_DESC, ED_STR, COUNT(*) AS Total, SUM(DEFAULTED) AS Defaults, ' + \\\n",
    "             'ROUND(SUM(DEFAULTED) * 100 / count(*)) AS PER_DEFAULT FROM CCDATA GROUP BY MARR_DESC, ED_STR ' + \\\n",
    "             'ORDER BY 1, 2').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+--------+-----------+\n",
      "|AVG_PAY_DUR|Total|Defaults|PER_DEFAULT|\n",
      "+-----------+-----+--------+-----------+\n",
      "|        0.0|  739|   270.0|       37.0|\n",
      "|        1.0|  239|   122.0|       51.0|\n",
      "|        2.0|   16|     9.0|       56.0|\n",
      "|        3.0|    6|     2.0|       33.0|\n",
      "+-----------+-----+--------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# PR#04 solution\n",
    "SpSession.sql('SELECT AVG_PAY_DUR, COUNT(*) AS Total, SUM(DEFAULTED) AS Defaults, ' + \\\n",
    "             'ROUND(SUM(DEFAULTED) * 100 / count(*)) AS PER_DEFAULT FROM CCDATA GROUP BY AVG_PAY_DUR ORDER BY 1').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Transform to a data frame for input to machine learning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Correlation to DEFAULTED for AGE is 0.516560220288\n",
      "Correlation to DEFAULTED for AVG_BILL_AMT is 0.186007563785\n",
      "Correlation to DEFAULTED for AVG_PAY_AMT is -0.16359608891\n",
      "Correlation to DEFAULTED for AVG_PAY_DUR is 0.115134607604\n",
      "Correlation to DEFAULTED for DEFAULTED is 1.0\n",
      "Correlation to DEFAULTED for EDUCATION is 0.11056265057\n",
      "Correlation to DEFAULTED for LIMIT_BAL is 0.10722031324\n",
      "Correlation to DEFAULTED for MARRIAGE is -0.228912872874\n",
      "Correlation to DEFAULTED for PER_PAID is -0.0268407104868\n",
      "Correlation to DEFAULTED for SEX is -0.0836518221502\n"
     ]
    }
   ],
   "source": [
    "# Perform first round correlation analysis\n",
    "for i in ccDf.columns:\n",
    "    if not(isinstance(ccDf.select(i).take(1)[0][0], unicode)):\n",
    "        print('Correlation to DEFAULTED for {} is {}'.format(i, ccDf.stat.corr('DEFAULTED', i)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.linalg import Vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def transformToLabeledPoint(row):\n",
    "    lp = (row['DEFAULTED'], Vectors.dense([row['AGE'], row['AVG_BILL_AMT'], row['AVG_PAY_AMT'], row['AVG_PAY_DUR'], \\\n",
    "                                          row['EDUCATION'], row['LIMIT_BAL'], row['MARRIAGE'], row['PER_PAID'], \\\n",
    "                                          row['SEX']]))\n",
    "    return lp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1.0,\n",
       "  DenseVector([60.0, -56043.1667, 57956.5, 0.0, 1.0, 480000.0, 1.0, -100.0, 1.0])),\n",
       " (1.0, DenseVector([60.0, 0.0, 0.0, 0.0, 1.0, 320000.0, 1.0, 0.0, 1.0])),\n",
       " (1.0, DenseVector([60.0, 0.0, 0.0, 0.0, 1.0, 230000.0, 1.0, 0.0, 1.0])),\n",
       " (1.0, DenseVector([60.0, 19763.0, 0.0, 1.0, 1.0, 500000.0, 1.0, 0.0, 1.0])),\n",
       " (1.0, DenseVector([60.0, -87.5, 0.0, 0.0, 1.0, 200000.0, 1.0, -0.0, 1.0]))]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ccLp = ccFinalDf.rdd.map(transformToLabeledPoint)\n",
    "ccLp.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------------------------------------------------------------+\n",
      "|label|features                                                          |\n",
      "+-----+------------------------------------------------------------------+\n",
      "|1.0  |[60.0,-56043.166666666664,57956.5,0.0,1.0,480000.0,1.0,-100.0,1.0]|\n",
      "|1.0  |[60.0,0.0,0.0,0.0,1.0,320000.0,1.0,0.0,1.0]                       |\n",
      "|1.0  |[60.0,0.0,0.0,0.0,1.0,230000.0,1.0,0.0,1.0]                       |\n",
      "|1.0  |[60.0,19763.0,0.0,1.0,1.0,500000.0,1.0,0.0,1.0]                   |\n",
      "|1.0  |[60.0,-87.5,0.0,0.0,1.0,200000.0,1.0,-0.0,1.0]                    |\n",
      "|1.0  |[60.0,125.0,0.0,1.0,1.0,50000.0,1.0,0.0,1.0]                      |\n",
      "|1.0  |[50.0,0.0,0.0,0.0,1.0,170000.0,1.0,0.0,1.0]                       |\n",
      "|1.0  |[50.0,1270.6666666666667,0.0,0.0,1.0,290000.0,1.0,0.0,1.0]        |\n",
      "|1.0  |[50.0,0.0,0.0,1.0,1.0,600000.0,1.0,0.0,1.0]                       |\n",
      "|1.0  |[50.0,81.83333333333333,0.0,1.0,1.0,200000.0,1.0,0.0,1.0]         |\n",
      "+-----+------------------------------------------------------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[label: double, features: vector]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ccNormDf = SpSession.createDataFrame(ccLp, ['label', 'features'])\n",
    "ccNormDf.select('*').show(10, truncate = False)\n",
    "ccNormDf.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+-------+\n",
      "|label|            features|indexed|\n",
      "+-----+--------------------+-------+\n",
      "|  1.0|[60.0,-56043.1666...|    1.0|\n",
      "|  1.0|[60.0,0.0,0.0,0.0...|    1.0|\n",
      "|  1.0|[60.0,0.0,0.0,0.0...|    1.0|\n",
      "|  1.0|[60.0,19763.0,0.0...|    1.0|\n",
      "|  1.0|[60.0,-87.5,0.0,0...|    1.0|\n",
      "+-----+--------------------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Indexing needed as pre-req\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "stringIndexer = StringIndexer(inputCol = 'label', outputCol = 'indexed')\n",
    "si_model = stringIndexer.fit(ccNormDf)\n",
    "td = si_model.transform(ccNormDf)\n",
    "td.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# split into training and testing data\n",
    "trainingData, testData = td.randomSplit([0.7, 0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "700"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "trainingData.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "300"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testData.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PR#05 Do predictions to predict defaults"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "evaluator = MulticlassClassificationEvaluator(predictionCol = 'prediction', labelCol = 'indexed', \n",
    "                                              metricName = 'accuracy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create the Dicision Tree model (with hyper-parameter tuning)\n",
    "paramGrid = ParamGridBuilder().addGrid(DecisionTreeClassifier.maxDepth, [4, 5, 6]).build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "dtClassifier = DecisionTreeClassifier(labelCol = 'indexed', featuresCol = 'features')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dtClassifier_cv = CrossValidator(estimator = DecisionTreeClassifier(), estimatorParamMaps = paramGrid,\n",
    "                                evaluator = MulticlassClassificationEvaluator(), numFolds = 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dtModel = dtClassifier_cv.fit(trainingData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+-----+--------------------+\n",
      "|prediction|indexed|label|            features|\n",
      "+----------+-------+-----+--------------------+\n",
      "|       0.0|    0.0|  0.0|[20.0,415.0,415.0...|\n",
      "|       0.0|    0.0|  0.0|[20.0,4689.5,4777...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12.83333333...|\n",
      "|       0.0|    0.0|  0.0|[30.0,4673.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,5737.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,6613.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,7506.5,4709...|\n",
      "|       0.0|    0.0|  0.0|[30.0,9349.5,1049...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12637.5,800...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12778.16666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,17449.0,130...|\n",
      "|       0.0|    0.0|  0.0|[30.0,17651.0,140...|\n",
      "|       0.0|    0.0|  0.0|[30.0,21045.33333...|\n",
      "|       0.0|    0.0|  0.0|[30.0,22045.16666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,27967.0,573...|\n",
      "|       0.0|    0.0|  0.0|[30.0,29458.5,214...|\n",
      "|       0.0|    0.0|  0.0|[30.0,41517.66666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,46249.5,291...|\n",
      "|       0.0|    0.0|  0.0|[30.0,57094.5,600...|\n",
      "|       0.0|    0.0|  0.0|[30.0,96540.5,370...|\n",
      "+----------+-------+-----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Predict on the test data\n",
    "predictions = dtModel.transform(testData)\n",
    "predictions.select('prediction', 'indexed', 'label', 'features').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Results of Decision Tree: 0.736666666667\n"
     ]
    }
   ],
   "source": [
    "print('Results of Decision Tree: {}'.format(evaluator.evaluate(predictions)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create the Random Forest model\n",
    "rfClassifier = RandomForestClassifier(labelCol = 'indexed', featuresCol = 'features')\n",
    "rfModel = rfClassifier.fit(trainingData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+-----+--------------------+\n",
      "|prediction|indexed|label|            features|\n",
      "+----------+-------+-----+--------------------+\n",
      "|       0.0|    0.0|  0.0|[20.0,415.0,415.0...|\n",
      "|       0.0|    0.0|  0.0|[20.0,4689.5,4777...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12.83333333...|\n",
      "|       0.0|    0.0|  0.0|[30.0,4673.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,5737.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,6613.166666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,7506.5,4709...|\n",
      "|       0.0|    0.0|  0.0|[30.0,9349.5,1049...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12637.5,800...|\n",
      "|       0.0|    0.0|  0.0|[30.0,12778.16666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,17449.0,130...|\n",
      "|       0.0|    0.0|  0.0|[30.0,17651.0,140...|\n",
      "|       0.0|    0.0|  0.0|[30.0,21045.33333...|\n",
      "|       0.0|    0.0|  0.0|[30.0,22045.16666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,27967.0,573...|\n",
      "|       0.0|    0.0|  0.0|[30.0,29458.5,214...|\n",
      "|       0.0|    0.0|  0.0|[30.0,41517.66666...|\n",
      "|       0.0|    0.0|  0.0|[30.0,46249.5,291...|\n",
      "|       0.0|    0.0|  0.0|[30.0,57094.5,600...|\n",
      "|       0.0|    0.0|  0.0|[30.0,96540.5,370...|\n",
      "+----------+-------+-----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Predict on the test data\n",
    "predictions = rfModel.transform(testData)\n",
    "predictions.select('prediction', 'indexed', 'label', 'features').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Results of Random Forest: 0.753333333333\n"
     ]
    }
   ],
   "source": [
    "print('Results of Random Forest: {}'.format(evaluator.evaluate(predictions)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Group data into 4 groups based on the said parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Filter only columns needed for clustering\n",
    "ccClustDf = ccFinalDf.select('SEX', 'EDUCATION', 'MARRIAGE', 'AGE', 'CUSTID')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>summary</th>\n",
       "      <th>SEX</th>\n",
       "      <th>EDUCATION</th>\n",
       "      <th>MARRIAGE</th>\n",
       "      <th>AGE</th>\n",
       "      <th>CUSTID</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>count</td>\n",
       "      <td>1000</td>\n",
       "      <td>1000</td>\n",
       "      <td>1000</td>\n",
       "      <td>1000</td>\n",
       "      <td>1000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>mean</td>\n",
       "      <td>1.591</td>\n",
       "      <td>1.769</td>\n",
       "      <td>1.608</td>\n",
       "      <td>35.42</td>\n",
       "      <td>500.5</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>stddev</td>\n",
       "      <td>0.49189527438364455</td>\n",
       "      <td>0.7212206036707212</td>\n",
       "      <td>0.5259397423779768</td>\n",
       "      <td>9.85498456217566</td>\n",
       "      <td>288.8194360957493</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>min</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>20.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>max</td>\n",
       "      <td>2.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>3.0</td>\n",
       "      <td>80.0</td>\n",
       "      <td>999</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  summary                  SEX           EDUCATION            MARRIAGE  \\\n",
       "0   count                 1000                1000                1000   \n",
       "1    mean                1.591               1.769               1.608   \n",
       "2  stddev  0.49189527438364455  0.7212206036707212  0.5259397423779768   \n",
       "3     min                  1.0                 1.0                 1.0   \n",
       "4     max                  2.0                 4.0                 3.0   \n",
       "\n",
       "                AGE             CUSTID  \n",
       "0              1000               1000  \n",
       "1             35.42              500.5  \n",
       "2  9.85498456217566  288.8194360957493  \n",
       "3              20.0                  1  \n",
       "4              80.0                999  "
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Do centering and scaling for the values\n",
    "summStats = ccClustDf.describe().toPandas()\n",
    "summStats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "meanValues = summStats.iloc[1, 1:].values.tolist()\n",
    "stdValues = summStats.iloc[2, 1:].values.tolist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "bcMeans = SpContext.broadcast(meanValues)\n",
    "bcStdDev = SpContext.broadcast(stdValues)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def centerAndScale(inRow):\n",
    "    global bcMeans\n",
    "    global bcStdDev\n",
    "    \n",
    "    meanArray = bcMeans.value\n",
    "    stdArray = bcStdDev.value\n",
    "    \n",
    "    retArray = []\n",
    "    for i in range(len(meanArray)):\n",
    "        retArray.append((float(inRow[i]) - float(meanArray[i])) / float(stdArray[i]))\n",
    "    return Row(CUSTID = inRow[4], features = Vectors.dense(retArray))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(CUSTID=u'103', features=DenseVector([-1.2015, -1.0662, -1.156, 2.4942, -1.3763])),\n",
       " Row(CUSTID=u'932', features=DenseVector([-1.2015, -1.0662, -1.156, 2.4942, 1.494])),\n",
       " Row(CUSTID=u'466', features=DenseVector([-1.2015, -1.0662, -1.156, 2.4942, -0.1195])),\n",
       " Row(CUSTID=u'35', features=DenseVector([-1.2015, -1.0662, -1.156, 2.4942, -1.6117])),\n",
       " Row(CUSTID=u'66', features=DenseVector([-1.2015, -1.0662, -1.156, 2.4942, -1.5044]))]"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ccMap = ccClustDf.rdd.map(centerAndScale)\n",
    "ccMap.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------------------------------------------------------------------------------------------------+\n",
      "|CUSTID|features                                                                                            |\n",
      "+------+----------------------------------------------------------------------------------------------------+\n",
      "|103   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,-1.376292417758966]  |\n",
      "|932   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,1.494013027076714]   |\n",
      "|466   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,-0.11945179474889138]|\n",
      "|35    |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,-1.611733636394462]  |\n",
      "|66    |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,-1.504400139663574]  |\n",
      "|553   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,2.4941693053828113,0.1817744702700521]  |\n",
      "|603   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,1.4794543723548166,0.35489301338438745] |\n",
      "|576   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,1.4794543723548166,0.26140900010264634] |\n",
      "|452   |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,1.4794543723548166,-0.1679249868209053] |\n",
      "|91    |[-1.2014752545458702,-1.0662479636412228,-1.156025968395157,1.4794543723548166,-1.4178408681064065] |\n",
      "+------+----------------------------------------------------------------------------------------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Create a data frame with the features\n",
    "ccFinalClustDf = SpSession.createDataFrame(ccMap)\n",
    "ccFinalClustDf.cache()\n",
    "ccFinalClustDf.show(10, truncate = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Perform clustering\n",
    "from pyspark.ml.clustering import KMeans\n",
    "kmeans = KMeans(k = 4, seed = 1)\n",
    "model = kmeans.fit(ccFinalClustDf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+----------+\n",
      "|CUSTID|            features|prediction|\n",
      "+------+--------------------+----------+\n",
      "|   103|[-1.2014752545458...|         3|\n",
      "|   932|[-1.2014752545458...|         3|\n",
      "|   466|[-1.2014752545458...|         3|\n",
      "|    35|[-1.2014752545458...|         3|\n",
      "|    66|[-1.2014752545458...|         3|\n",
      "|   553|[-1.2014752545458...|         3|\n",
      "|   603|[-1.2014752545458...|         3|\n",
      "|   576|[-1.2014752545458...|         3|\n",
      "|   452|[-1.2014752545458...|         3|\n",
      "|    91|[-1.2014752545458...|         3|\n",
      "|   406|[-1.2014752545458...|         3|\n",
      "|   451|[-1.2014752545458...|         3|\n",
      "|   395|[-1.2014752545458...|         3|\n",
      "|    18|[-1.2014752545458...|         3|\n",
      "|   917|[-1.2014752545458...|         3|\n",
      "|   260|[-1.2014752545458...|         3|\n",
      "|   724|[-1.2014752545458...|         3|\n",
      "|   686|[-1.2014752545458...|         3|\n",
      "|   207|[-1.2014752545458...|         3|\n",
      "|   627|[-1.2014752545458...|         3|\n",
      "+------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "predictions = model.transform(ccFinalClustDf)\n",
    "predictions.select('*').show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}