Lab 3 - Machine Learning Student.ipynb

{
    "cells": [
        {
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "# Machine learning - Student notebook\n   \n   - Dataset, RDD APIs\n   - A dataframe is nothing but a table with columns, rows and headers\n   - In this notebook, we will work through a dataset to demostrate spark's SQL-like abilities. We will also look at its machine learning capabilities in context of this dataset.\n    \n### Dataset\n- The chosen dataset is that of Breast Cancer. This set is collected from digitized image of a fine needle aspirate (FNA) of a breast mass. For more information about this dataset, please visit https://www.kaggle.com/uciml/breast-cancer-wisconsin-data\n- Each row contains information about a single breast mass along with a diagnosis of malignant/benign for this mass\n    \n### Problem statement\n- We need to learn patterns describing malignant and benign masses and need to be able to place any future samples in either of these buckets.  \n- Thus, the problem is a classification one where given a set of probable results, we need to pick one result with confidance. \n    \n### Solution\n   We will go through the following steps:\n   - Read data (Spark SQL)\n   - Feature Engineering (Accumulators, Broadcasters, ml/mllib APIs)\n   - Data Visualizations (PixieDust, Seaborn)\n   - Modeling (Spark ML)\n   - Evaluation and prediction (Spark ML)\n   - Deployment (Watson ML repository)"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "from IPython.display import display\nfrom IPython.core.display import HTML ", 
            "outputs": [], 
            "execution_count": 1
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "!pip install seaborn", 
            "outputs": [
                {
                    "text": "Requirement already satisfied: seaborn in /gpfs/global_fs01/sym_shared/YPProdSpark/user/s280-a46e86a84e2274-95b1885704d4/.local/lib/python2.7/site-packages\r\n", 
                    "output_type": "stream", 
                    "name": "stdout"
                }
            ], 
            "execution_count": 2
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Machine Learning process ###\n\n<center><img width=\"600px\" height=\"600px\" src=\"https://raw.githubusercontent.com/martinkearn/Content/master/Blogs/Images/MLProcess.PNG\"></center>"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Get dataset\n!wget https://raw.githubusercontent.com/joshishwetha/dsx-spark/master/data.csv"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "!wget https://raw.githubusercontent.com/joshishwetha/dsx-spark/master/data.csv", 
            "outputs": [
                {
                    "text": "--2017-04-26 17:17:59--  https://raw.githubusercontent.com/joshishwetha/dsx-spark/master/data.csv\nResolving raw.githubusercontent.com (raw.githubusercontent.com)... 151.101.48.133\nConnecting to raw.githubusercontent.com (raw.githubusercontent.com)|151.101.48.133|:443... connected.\nHTTP request sent, awaiting response... 200 OK\nLength: 125139 (122K) [text/plain]\nSaving to: \u2018data.csv.7\u2019\n\n100%[======================================>] 125,139     --.-K/s   in 0.006s  \n\n2017-04-26 17:17:59 (19.3 MB/s) - \u2018data.csv.7\u2019 saved [125139/125139]\n\n", 
                    "output_type": "stream", 
                    "name": "stdout"
                }
            ], 
            "execution_count": 10
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Read dataset\n\n#### copy and paste the following code\n    data = spark.read.csv('data.csv',inferSchema='true',header='true')\n    data = data.drop('_c32')"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "# data = spark.read.csv('data.csv',inferSchema='true',header='true')\ndata = data.drop('_c32')", 
            "outputs": [], 
            "execution_count": 14
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## Data Exploration ###"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "<center><img width=\"450px\" height=\"450px\" margin=\"auto\" src=\"https://img.clipartfest.com/f92b25f421c985eed2ccb12cdb4cbf54_vector-clip-art-cartoon-safari-kids-cartoon-clipart_800-557.jpeg\"></center>"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Basic operations on dataframe \n- To view contents: df.show(), df.take(n)\n- To transform columns: df.withColumn (\"column_name\",\"transformation\")\n- To rename columns: df.withColumnRenamed(\"old_name\",\"new_name\")"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Print first 3 rows of the input dataset ###\n\n    data.toPandas()[:3]"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "data.toPandas()[:3]", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "         id diagnosis  radius_mean  texture_mean  perimeter_mean  area_mean  \\\n0    842302         M        17.99         10.38           122.8       1001   \n1    842517         M        20.57         17.77           132.9       1326   \n2  84300903         M        19.69         21.25           130.0       1203   \n\n   smoothness_mean  compactness_mean  concavity_mean  concave points_mean  \\\n0          0.11840           0.27760          0.3001              0.14710   \n1          0.08474           0.07864          0.0869              0.07017   \n2          0.10960           0.15990          0.1974              0.12790   \n\n            ...             radius_worst  texture_worst  perimeter_worst  \\\n0           ...                    25.38          17.33            184.6   \n1           ...                    24.99          23.41            158.8   \n2           ...                    23.57          25.53            152.5   \n\n   area_worst  smoothness_worst  compactness_worst  concavity_worst  \\\n0        2019            0.1622             0.6656           0.7119   \n1        1956            0.1238             0.1866           0.2416   \n2        1709            0.1444             0.4245           0.4504   \n\n   concave points_worst  symmetry_worst  fractal_dimension_worst  \n0                0.2654          0.4601                  0.11890  \n1                0.1860          0.2750                  0.08902  \n2                0.2430          0.3613                  0.08758  \n\n[3 rows x 32 columns]", 
                        "text/html": "<div>\n<table border=\"1\" class=\"dataframe\">\n  <thead>\n    <tr style=\"text-align: right;\">\n      <th></th>\n      <th>id</th>\n      <th>diagnosis</th>\n      <th>radius_mean</th>\n      <th>texture_mean</th>\n      <th>perimeter_mean</th>\n      <th>area_mean</th>\n      <th>smoothness_mean</th>\n      <th>compactness_mean</th>\n      <th>concavity_mean</th>\n      <th>concave points_mean</th>\n      <th>...</th>\n      <th>radius_worst</th>\n      <th>texture_worst</th>\n      <th>perimeter_worst</th>\n      <th>area_worst</th>\n      <th>smoothness_worst</th>\n      <th>compactness_worst</th>\n      <th>concavity_worst</th>\n      <th>concave points_worst</th>\n      <th>symmetry_worst</th>\n      <th>fractal_dimension_worst</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>842302</td>\n      <td>M</td>\n      <td>17.99</td>\n      <td>10.38</td>\n      <td>122.8</td>\n      <td>1001</td>\n      <td>0.11840</td>\n      <td>0.27760</td>\n      <td>0.3001</td>\n      <td>0.14710</td>\n      <td>...</td>\n      <td>25.38</td>\n      <td>17.33</td>\n      <td>184.6</td>\n      <td>2019</td>\n      <td>0.1622</td>\n      <td>0.6656</td>\n      <td>0.7119</td>\n      <td>0.2654</td>\n      <td>0.4601</td>\n      <td>0.11890</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>842517</td>\n      <td>M</td>\n      <td>20.57</td>\n      <td>17.77</td>\n      <td>132.9</td>\n      <td>1326</td>\n      <td>0.08474</td>\n      <td>0.07864</td>\n      <td>0.0869</td>\n      <td>0.07017</td>\n      <td>...</td>\n      <td>24.99</td>\n      <td>23.41</td>\n      <td>158.8</td>\n      <td>1956</td>\n      <td>0.1238</td>\n      <td>0.1866</td>\n      <td>0.2416</td>\n      <td>0.1860</td>\n      <td>0.2750</td>\n      <td>0.08902</td>\n    </tr>\n    <tr>\n      <th>2</th>\n      <td>84300903</td>\n      <td>M</td>\n      <td>19.69</td>\n      <td>21.25</td>\n      <td>130.0</td>\n      <td>1203</td>\n      <td>0.10960</td>\n      <td>0.15990</td>\n      <td>0.1974</td>\n      <td>0.12790</td>\n      <td>...</td>\n      <td>23.57</td>\n      <td>25.53</td>\n      <td>152.5</td>\n      <td>1709</td>\n      <td>0.1444</td>\n      <td>0.4245</td>\n      <td>0.4504</td>\n      <td>0.2430</td>\n      <td>0.3613</td>\n      <td>0.08758</td>\n    </tr>\n  </tbody>\n</table>\n<p>3 rows \u00d7 32 columns</p>\n</div>"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 15
                }
            ], 
            "execution_count": 15
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Check datatypes\ndata.dtypes"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "data.dtypes", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "[('id', 'int'),\n ('diagnosis', 'string'),\n ('radius_mean', 'double'),\n ('texture_mean', 'double'),\n ('perimeter_mean', 'double'),\n ('area_mean', 'double'),\n ('smoothness_mean', 'double'),\n ('compactness_mean', 'double'),\n ('concavity_mean', 'double'),\n ('concave points_mean', 'double'),\n ('symmetry_mean', 'double'),\n ('fractal_dimension_mean', 'double'),\n ('radius_se', 'double'),\n ('texture_se', 'double'),\n ('perimeter_se', 'double'),\n ('area_se', 'double'),\n ('smoothness_se', 'double'),\n ('compactness_se', 'double'),\n ('concavity_se', 'double'),\n ('concave points_se', 'double'),\n ('symmetry_se', 'double'),\n ('fractal_dimension_se', 'double'),\n ('radius_worst', 'double'),\n ('texture_worst', 'double'),\n ('perimeter_worst', 'double'),\n ('area_worst', 'double'),\n ('smoothness_worst', 'double'),\n ('compactness_worst', 'double'),\n ('concavity_worst', 'double'),\n ('concave points_worst', 'double'),\n ('symmetry_worst', 'double'),\n ('fractal_dimension_worst', 'double')]"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 16
                }
            ], 
            "execution_count": 16
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Make sure the data types are correct ####\n\n    from pyspark.sql.types import FloatType, StringType, IntegerType, DoubleType, ArrayType\n\n    for col in data.columns:\n        if col not in ['id','diagnosis']:\n            data = data.withColumn(col,data[col].cast(FloatType()))"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "from pyspark.sql.types import FloatType, StringType, IntegerType, DoubleType, ArrayType\n\nfor col in data.columns:\n    if col not in ['id','diagnosis']:\n        data = data.withColumn(col,data[col].cast(FloatType()))", 
            "outputs": [], 
            "execution_count": 17
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Replace null values with 0\n    data = data.na.fill(0)"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "data = data.na.fill(0)", 
            "outputs": [], 
            "execution_count": 18
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### SQL\n\n#### Register dataframe as a temp table to query from (write sql on dataframes)\n    data.registerTempTable('cancer_data')"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "", 
            "outputs": [], 
            "execution_count": null
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Querying data\n- NOTE that the returned object is another Dataframe\n- One nice feature of the notebooks and python is that we can show it in a table via Pandas\n- Remember to perform an action to get your results (sql queries are also transformations :))"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "  \n    query = \"\"\"\n    select\n        diagnosis ,\n        count(1) as diagnosis_count\n    from cancer_data\n    group by diagnosis \n    \"\"\"\n    spark.sql(query).toPandas()\n"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "", 
            "outputs": [], 
            "execution_count": null
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## Feature Engineering\n   - Dataset = Features + target\n   - Transforming features (categorical to numeric, continous to bins, scaling, normalization etc...)\n   - Selecting a subset of columns for wide datasets\n   - Exploding columns to make additional (synthetic features) for small datasets"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Feature set is numeric\n\n##### Step 1: Make sure your target is numeric\n- \"target\" is categorical (malignant/benign)\n- Our encoding: 1 = malignant(M), 0 = Benign(B) using sql like \"when\" statement"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Review your data\n#### Sample your data to view what it looks like before we make changes to the target column\n\n    data.select('diagnosis','radius_se').show(3)"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "data.select('diagnosis','radius_se').show(3)", 
            "outputs": [
                {
                    "text": "+---------+---------+\n|diagnosis|radius_se|\n+---------+---------+\n|        M|    1.095|\n|        M|   0.5435|\n|        M|   0.7456|\n+---------+---------+\nonly showing top 3 rows\n\n", 
                    "output_type": "stream", 
                    "name": "stdout"
                }
            ], 
            "execution_count": 19
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Make label numeric\n\n##### copy and paste to run code\n    import  pyspark.sql.functions as F\n\n    data = data.withColumn('diagnosis',F.when(data.diagnosis=='M',1).otherwise(0))\n    data.toPandas().head(3)"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "import  pyspark.sql.functions as F\n\ndata = data.withColumn('diagnosis',F.when(data.diagnosis=='M',1).otherwise(0))\ndata.toPandas().head(3)", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "         id  diagnosis  radius_mean  texture_mean  perimeter_mean  area_mean  \\\n0    842302          1    17.990000         10.38      122.800003       1001   \n1    842517          1    20.570000         17.77      132.899994       1326   \n2  84300903          1    19.690001         21.25      130.000000       1203   \n\n   smoothness_mean  compactness_mean  concavity_mean  concave points_mean  \\\n0          0.11840           0.27760          0.3001              0.14710   \n1          0.08474           0.07864          0.0869              0.07017   \n2          0.10960           0.15990          0.1974              0.12790   \n\n            ...             radius_worst  texture_worst  perimeter_worst  \\\n0           ...                25.379999      17.330000       184.600006   \n1           ...                24.990000      23.410000       158.800003   \n2           ...                23.570000      25.530001       152.500000   \n\n   area_worst  smoothness_worst  compactness_worst  concavity_worst  \\\n0        2019            0.1622             0.6656           0.7119   \n1        1956            0.1238             0.1866           0.2416   \n2        1709            0.1444             0.4245           0.4504   \n\n   concave points_worst  symmetry_worst  fractal_dimension_worst  \n0                0.2654          0.4601                  0.11890  \n1                0.1860          0.2750                  0.08902  \n2                0.2430          0.3613                  0.08758  \n\n[3 rows x 32 columns]", 
                        "text/html": "<div>\n<table border=\"1\" class=\"dataframe\">\n  <thead>\n    <tr style=\"text-align: right;\">\n      <th></th>\n      <th>id</th>\n      <th>diagnosis</th>\n      <th>radius_mean</th>\n      <th>texture_mean</th>\n      <th>perimeter_mean</th>\n      <th>area_mean</th>\n      <th>smoothness_mean</th>\n      <th>compactness_mean</th>\n      <th>concavity_mean</th>\n      <th>concave points_mean</th>\n      <th>...</th>\n      <th>radius_worst</th>\n      <th>texture_worst</th>\n      <th>perimeter_worst</th>\n      <th>area_worst</th>\n      <th>smoothness_worst</th>\n      <th>compactness_worst</th>\n      <th>concavity_worst</th>\n      <th>concave points_worst</th>\n      <th>symmetry_worst</th>\n      <th>fractal_dimension_worst</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>842302</td>\n      <td>1</td>\n      <td>17.990000</td>\n      <td>10.38</td>\n      <td>122.800003</td>\n      <td>1001</td>\n      <td>0.11840</td>\n      <td>0.27760</td>\n      <td>0.3001</td>\n      <td>0.14710</td>\n      <td>...</td>\n      <td>25.379999</td>\n      <td>17.330000</td>\n      <td>184.600006</td>\n      <td>2019</td>\n      <td>0.1622</td>\n      <td>0.6656</td>\n      <td>0.7119</td>\n      <td>0.2654</td>\n      <td>0.4601</td>\n      <td>0.11890</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>842517</td>\n      <td>1</td>\n      <td>20.570000</td>\n      <td>17.77</td>\n      <td>132.899994</td>\n      <td>1326</td>\n      <td>0.08474</td>\n      <td>0.07864</td>\n      <td>0.0869</td>\n      <td>0.07017</td>\n      <td>...</td>\n      <td>24.990000</td>\n      <td>23.410000</td>\n      <td>158.800003</td>\n      <td>1956</td>\n      <td>0.1238</td>\n      <td>0.1866</td>\n      <td>0.2416</td>\n      <td>0.1860</td>\n      <td>0.2750</td>\n      <td>0.08902</td>\n    </tr>\n    <tr>\n      <th>2</th>\n      <td>84300903</td>\n      <td>1</td>\n      <td>19.690001</td>\n      <td>21.25</td>\n      <td>130.000000</td>\n      <td>1203</td>\n      <td>0.10960</td>\n      <td>0.15990</td>\n      <td>0.1974</td>\n      <td>0.12790</td>\n      <td>...</td>\n      <td>23.570000</td>\n      <td>25.530001</td>\n      <td>152.500000</td>\n      <td>1709</td>\n      <td>0.1444</td>\n      <td>0.4245</td>\n      <td>0.4504</td>\n      <td>0.2430</td>\n      <td>0.3613</td>\n      <td>0.08758</td>\n    </tr>\n  </tbody>\n</table>\n<p>3 rows \u00d7 32 columns</p>\n</div>"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 20
                }
            ], 
            "execution_count": 20
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Rename column from \"diagnosis\" to \"label\"\n\n    data = data.withColumnRenamed('diagnosis','label')"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "# data = data.withColumnRenamed('diagnosis','label')\ndata.columns", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "['id',\n 'label',\n 'radius_mean',\n 'texture_mean',\n 'perimeter_mean',\n 'area_mean',\n 'smoothness_mean',\n 'compactness_mean',\n 'concavity_mean',\n 'concave points_mean',\n 'symmetry_mean',\n 'fractal_dimension_mean',\n 'radius_se',\n 'texture_se',\n 'perimeter_se',\n 'area_se',\n 'smoothness_se',\n 'compactness_se',\n 'concavity_se',\n 'concave points_se',\n 'symmetry_se',\n 'fractal_dimension_se',\n 'radius_worst',\n 'texture_worst',\n 'perimeter_worst',\n 'area_worst',\n 'smoothness_worst',\n 'compactness_worst',\n 'concavity_worst',\n 'concave points_worst',\n 'symmetry_worst',\n 'fractal_dimension_worst']"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 23
                }
            ], 
            "execution_count": 23
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "***********************************************************************************\n\n###   Detour  \n- Feature engineering is a major task and often involves complicated scripts\n\nWhat happens when\n   - Have custom script based on other datasets/static variables?\n   - I want to port over a current script to spark ?"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Accumulators, Broadcasters\n- One of the most basic things in scripts are variables (data and metadata). \n- Data vars = DataFrames/RDDs in spark\n- Metadata vars = Accumulators/broadcasters \n   \n- Metadata vars are not straightforward in the distributed world\n\n\n##### Accumulators\n   - Global variables which can be written into\n   \n##### Broadcasters\n   - Global variables to be read from \n************************************************************************************************"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Lets run a simple accumulator\n- Problem: Count the number of benign and malignant cases"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "def count_labels(row):\n    global benign, malignant\n    if row.label==1:\n        malignant.add(1)\n    else:\n        benign.add(1)", 
            "outputs": [], 
            "execution_count": 1
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### copy and run this code\n    benign = sc.accumulator(0)\n    malignant = sc.accumulator(0)\n\n    data.rdd.foreach(count_labels)\n\n    benign.value, malignant.value"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "benign = sc.accumulator(0)\nmalignant = sc.accumulator(0)\n\ndata.rdd.foreach(count_labels)\n\nbenign.value, malignant.value", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "(357, 212)"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 24
                }
            ], 
            "execution_count": 24
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Broadcast the data variable so that each executor has its own version of it\n   - This way, it does need not be shipped to the executor with every call\n   - Saves network bandwidth"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Collect statistics for your dataset ###\n- Collect statistics for mean/std\n- store it in a variable called \"df\"\n\n\n        df = data.describe()\n        df.toPandas()"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "df = data.describe()\ndf.toPandas()", 
            "outputs": [
                {
                    "metadata": {}, 
                    "data": {
                        "text/plain": "  summary                    id                label         radius_mean  \\\n0   count                   569                  569                 569   \n1    mean  3.0371831432337433E7  0.37258347978910367  14.127291743072348   \n2  stddev  1.2502058561222367E8   0.4839179564031687   3.524048812967195   \n3     min                  8670                    0               6.981   \n4     max             911320502                    1               28.11   \n\n         texture_mean      perimeter_mean           area_mean  \\\n0                 569                 569                 569   \n1  19.289648528677297   91.96903329993384    654.889103814043   \n2   4.301035792275387  24.298980946187076  351.91412886139733   \n3                9.71               43.79               143.5   \n4               39.28               188.5              2501.0   \n\n        smoothness_mean     compactness_mean       concavity_mean  \\\n0                   569                  569                  569   \n1   0.09636028129312821  0.10434098429781481  0.08879931578830029   \n2  0.014064128011679866  0.05281275807458227  0.07971980885275734   \n3               0.05263              0.01938                  0.0   \n4                0.1634               0.3454               0.4268   \n\n            ...                  radius_worst      texture_worst  \\\n0           ...                           569                569   \n1           ...            16.269189776770887  25.67722316534113   \n2           ...            4.8332415912724365  6.146257611231105   \n3           ...                          7.93              12.02   \n4           ...                         36.04              49.54   \n\n      perimeter_worst         area_worst      smoothness_worst  \\\n0                 569                569                   569   \n1  107.26121279644421  880.5831290514901   0.13236859435565862   \n2  33.602542264508905  569.3569923849643  0.022832429559187094   \n3               50.41              185.2               0.07117   \n4               251.2             4254.0                0.2226   \n\n     compactness_worst      concavity_worst concave points_worst  \\\n0                  569                  569                  569   \n1  0.25426504394016597   0.2721884833807977  0.11460622294146325   \n2  0.15733648854662943  0.20862428007810724  0.06573234105890066   \n3              0.02729                  0.0                  0.0   \n4                1.058                1.252                0.291   \n\n        symmetry_worst fractal_dimension_worst  \n0                  569                     569  \n1   0.2900755708948799     0.08394581713895387  \n2  0.06186746818484165     0.01806126727675421  \n3               0.1565                 0.05504  \n4               0.6638                  0.2075  \n\n[5 rows x 33 columns]", 
                        "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>id</th>\n      <th>label</th>\n      <th>radius_mean</th>\n      <th>texture_mean</th>\n      <th>perimeter_mean</th>\n      <th>area_mean</th>\n      <th>smoothness_mean</th>\n      <th>compactness_mean</th>\n      <th>concavity_mean</th>\n      <th>...</th>\n      <th>radius_worst</th>\n      <th>texture_worst</th>\n      <th>perimeter_worst</th>\n      <th>area_worst</th>\n      <th>smoothness_worst</th>\n      <th>compactness_worst</th>\n      <th>concavity_worst</th>\n      <th>concave points_worst</th>\n      <th>symmetry_worst</th>\n      <th>fractal_dimension_worst</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>count</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>...</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n      <td>569</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>mean</td>\n      <td>3.0371831432337433E7</td>\n      <td>0.37258347978910367</td>\n      <td>14.127291743072348</td>\n      <td>19.289648528677297</td>\n      <td>91.96903329993384</td>\n      <td>654.889103814043</td>\n      <td>0.09636028129312821</td>\n      <td>0.10434098429781481</td>\n      <td>0.08879931578830029</td>\n      <td>...</td>\n      <td>16.269189776770887</td>\n      <td>25.67722316534113</td>\n      <td>107.26121279644421</td>\n      <td>880.5831290514901</td>\n      <td>0.13236859435565862</td>\n      <td>0.25426504394016597</td>\n      <td>0.2721884833807977</td>\n      <td>0.11460622294146325</td>\n      <td>0.2900755708948799</td>\n      <td>0.08394581713895387</td>\n    </tr>\n    <tr>\n      <th>2</th>\n      <td>stddev</td>\n      <td>1.2502058561222367E8</td>\n      <td>0.4839179564031687</td>\n      <td>3.524048812967195</td>\n      <td>4.301035792275387</td>\n      <td>24.298980946187076</td>\n      <td>351.91412886139733</td>\n      <td>0.014064128011679866</td>\n      <td>0.05281275807458227</td>\n      <td>0.07971980885275734</td>\n      <td>...</td>\n      <td>4.8332415912724365</td>\n      <td>6.146257611231105</td>\n      <td>33.602542264508905</td>\n      <td>569.3569923849643</td>\n      <td>0.022832429559187094</td>\n      <td>0.15733648854662943</td>\n      <td>0.20862428007810724</td>\n      <td>0.06573234105890066</td>\n      <td>0.06186746818484165</td>\n      <td>0.01806126727675421</td>\n    </tr>\n    <tr>\n      <th>3</th>\n      <td>min</td>\n      <td>8670</td>\n      <td>0</td>\n      <td>6.981</td>\n      <td>9.71</td>\n      <td>43.79</td>\n      <td>143.5</td>\n      <td>0.05263</td>\n      <td>0.01938</td>\n      <td>0.0</td>\n      <td>...</td>\n      <td>7.93</td>\n      <td>12.02</td>\n      <td>50.41</td>\n      <td>185.2</td>\n      <td>0.07117</td>\n      <td>0.02729</td>\n      <td>0.0</td>\n      <td>0.0</td>\n      <td>0.1565</td>\n      <td>0.05504</td>\n    </tr>\n    <tr>\n      <th>4</th>\n      <td>max</td>\n      <td>911320502</td>\n      <td>1</td>\n      <td>28.11</td>\n      <td>39.28</td>\n      <td>188.5</td>\n      <td>2501.0</td>\n      <td>0.1634</td>\n      <td>0.3454</td>\n      <td>0.4268</td>\n      <td>...</td>\n      <td>36.04</td>\n      <td>49.54</td>\n      <td>251.2</td>\n      <td>4254.0</td>\n      <td>0.2226</td>\n      <td>1.058</td>\n      <td>1.252</td>\n      <td>0.291</td>\n      <td>0.6638</td>\n      <td>0.2075</td>\n    </tr>\n  </tbody>\n</table>\n<p>5 rows \u00d7 33 columns</p>\n</div>"
                    }, 
                    "output_type": "execute_result", 
                    "execution_count": 27
                }
            ], 
            "execution_count": 27
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Calculate correlations of each columns w.r.t the target\n- Change to do this in spark\n\n\n\n#### Select a subset of columns\n- Remove columns conveying very little information\n   - Zero variance columns\n   - Low correlation columns"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "df = data.toPandas()\ncol_corr = sorted(df.corr()['label'].to_dict().items(),key=lambda x:x[1],reverse=True)\n\nfor col in col_corr[-5:]:\n    if col[0]!='id':\n        data = data.drop(col[0])", 
            "outputs": [], 
            "execution_count": 28
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "- Remove columns with low correlation, such columns contribute to very little information\n- Caution: Be wary of multicolinearity (\"leaky\" columns)"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## Lets do some graphs!\n\n###  Here, we look at the columns and their correlations with the target variable.\n- We use PixieDust (charting library) for quick charts\n- Also takes RDDs as inputs, most of the other charting librabries take pandas dataframes as inputs\n\n\n##### copy and run code from here\n    import pixiedust\n    import pandas as pd\n    \n    df = pd.DataFrame(col_corr)\n    df = df.dropna()\n    df.columns = ['name','correlation']\n    viz_spark_df = sqlContext.createDataFrame(df)\n\n   \n    display(viz_spark_df)"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "pixiedust": {
                    "displayParams": {
                        "handlerId": "barChart", 
                        "rowCount": "100", 
                        "aggregation": "SUM", 
                        "keyFields": "name", 
                        "valueFields": "correlation"
                    }
                }
            }, 
            "source": "import pixiedust\nimport pandas as pd\n\ndf = pd.DataFrame(col_corr)\ndf = df.dropna()\ndf.columns = ['name','correlation']\nviz_spark_df = sqlContext.createDataFrame(df)\n\n\ndisplay(viz_spark_df)", 
            "outputs": [], 
            "execution_count": 29
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Visualize correlations ####\n - Useful for datasets with small number of columns\n - A heatmap is useful to visualize how variables are related to each other (and not just the target)\n - Very useful while doing NLP applications to visualize similarity between documents (after using TFIDF)\n "
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "import matplotlib.pyplot as plt\nimport seaborn as sns\n\nfrom pyspark.mllib.stat import Statistics\n\nplt.style.use('ggplot')\nvalues = data.rdd.map(lambda x:list(x.asDict().values()))\ncorr_values = Statistics.corr(values)\nnames = data.rdd.map(lambda x:list(x.asDict().keys())).first()\n\nplt.figure(figsize=(10,10))\nsns.set_style(\"darkgrid\")\nsns.heatmap(corr_values,xticklabels=names,yticklabels=names,square=True,vmin=0, vmax=1,\n                cmap=\"YlGnBu\")", 
            "outputs": [], 
            "execution_count": 2
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## We are all set to do some machine learning!\n\n<center><img width=\"350px\" height=\"350px\" margin=\"auto\" src=\"http://www.clipartkid.com/images/127/cartoon-explorer-characters-vectors-6cbSmI-clipart.jpg\"></center>"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## Module layout\n##### It divides into two packages:\n- spark.mllib contains the original API built on top of RDDs.\n- spark.ml provides higher-level API built on top of DataFrames for constructing ML pipelines.\n- Using spark.ml is recommended because with DataFrames the API is more versatile and flexible. But we will keep supporting spark.mllib along with the development of spark.ml. Users should be comfortable using spark.mllib features and expect more features coming.\n\n    \n    \n#### A common receipe\n- Make sure your data is numeric \n- Collect feature columns\n- Define transformations (Vector Indexer, Encoder)\n- Assemble columns into a single Vector column (VectorAssembler)\n- Split data into test and train\n- Define pipelines and estimators"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Machine learning Pipelines\n- Easy to use API\n- Define stages to transform your dataset\n- Select any estimator (model) for prediction \n- Each pipeline will model a single estimator onto your model\n\n\n<center><img src=\"https://image.slidesharecdn.com/mlwithapachespark-2-160818115257/95/introduction-to-ml-with-apache-spark-mllib-47-638.jpg?cb=1490306278\" width=\"500px\" height=\"500px\"></center>"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### What are we trying to again???"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### PCA  - Dimensionality Reduction \n - Very handy for visualization!"
        }, 
        {
            "cell_type": "code", 
            "metadata": {}, 
            "source": "###### PCA Visualize data ######\nfrom pyspark.ml.feature import PCA\nfrom pyspark.ml.linalg import Vectors\nfrom pyspark.sql.functions import UserDefinedFunction\nfrom pyspark.ml.feature import VectorAssembler \n\nfeature_cols = list(filter(lambda x:x not in ['id','label','dummy'],data.columns))\nassembler = VectorAssembler(inputCols=feature_cols,outputCol='features')\ndf = assembler.transform(data)\n\npca = PCA(k=2, inputCol=\"features\", outputCol=\"pcaFeatures\")\nmodel = pca.fit(df)\n\npca_result = model.transform(df)\nresult = pca_result.select(\"pcaFeatures\")", 
            "outputs": [], 
            "execution_count": 2
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "# import regex\n\n# def g(point,co_ord):\n#     if co_ord=='x':\n#         value = regex.split('\\s+',point)[0][1:]\n#     else:\n#         value =  regex.split('\\s+',point)[1][:-1]\n#     return value\n\n# get_x_coord = UserDefinedFunction(lambda point:g(point,'x'), StringType())\n# get_y_coord = UserDefinedFunction(lambda point:g(point,'y'), StringType())\n# convert_to_string = UserDefinedFunction(lambda point:str(point.toArray()),StringType())\n\n# pca_result = pca_result.withColumn('pcaFeatures_string',convert_to_string(pca_result['pcaFeatures']))\n# pca_result = pca_result.withColumn('x_coord',get_x_coord(pca_result['pcaFeatures_string']))\\\n#                        .withColumn('y_coord',get_y_coord(pca_result['pcaFeatures_string']))\\\n                      \n# pca_result = pca_result.withColumn('x_coord',pca_result.x_coord.cast(FloatType()))\\\n#                        .withColumn('y_coord',pca_result.y_coord.cast(FloatType()))\n    \n# viz_df = pca_result.select('x_coord','y_coord','label').toPandas()\n# a = sns.lmplot(x='x_coord',y='y_coord',hue='label',data=viz_df,\n#            fit_reg=False,palette={1:'red',0:'green'},size=6,aspect=1)", 
            "outputs": [], 
            "execution_count": 3
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Prepare the dataset for machine learning\n    - Split data between test and train (70/30 split)\n    - Apply logistic regression and decision tree classifiers"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Convert 'label' to DoubleType()"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "data = data.withColumn('label',data['label'].cast(DoubleType()))", 
            "outputs": [], 
            "execution_count": 4
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Split dataset into test and train sets (70/30) splits"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "train, test = data.randomSplit([0.7,0.3])", 
            "outputs": [], 
            "execution_count": 5
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Count number of records in test and train sets"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "train.count(), test.count()", 
            "outputs": [], 
            "execution_count": 6
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Define Assembler to combine feature into a single vetor (feature vector! :) )"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "from pyspark.ml.pipeline import Pipeline\nfrom pyspark.ml.linalg import Vectors #linear algebra package, has matrices, arrays, Vectors (dense and sparse)\nfrom pyspark.ml.feature import VectorAssembler \n\nfeature_cols = list(filter(lambda x:x not in ['id','label'],data.columns))\nassembler = VectorAssembler(inputCols=feature_cols,outputCol='features')", 
            "outputs": [], 
            "execution_count": 39
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "## Classification Pipelines \n- Logistic Regression\n- Tree based classifier\n\n<center><img src=\"https://codesachin.files.wordpress.com/2015/08/linearly_separable_4.png\"></center>"
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Tree classifier ###"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "from pyspark.ml.classification import DecisionTreeClassifier\n# define estimator and fit data\nestimator = DecisionTreeClassifier()\npipeline = Pipeline(stages=[assembler,estimator])\ntree_model = pipeline.fit(train)\n\n# get results\ndt_results = tree_model.transform(test)", 
            "outputs": [], 
            "execution_count": 7
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Vizualise results of a tree based classifier"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "### Visualize the results of tree classifier ###\npca_subset = pca_result.select('ID','features','x_coord','y_coord')\nviz_df = pca_subset.join(results,on=['ID','features']).select('x_coord','y_coord','prediction','label')\nviz_df = viz_df.dropna(how='any',subset=['x_coord','y_coord']).toPandas()\n\na = sns.lmplot(x='x_coord',y='y_coord',hue='prediction',col='label',data=viz_df,\n           fit_reg=False,palette={1:'red',0:'green'},size=6,aspect=1)", 
            "outputs": [], 
            "execution_count": 8
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "##### Logistic regression #####"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "from pyspark.ml.classification import LogisticRegression\n\n#define estimator and fit data\nestimator = LogisticRegression()\npipeline = Pipeline(stages=[assembler,estimator])\nlr_model = pipeline.fit(train)\n\n#get results\nlr_results = lr_model.transform(test)", 
            "outputs": [], 
            "execution_count": 9
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "lr_results.select('id','features','label','probability','prediction').show(5)", 
            "outputs": [], 
            "execution_count": 10
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "### Visualize results from logistic regression ###\npca_subset = pca_result.select('id','features','x_coord','y_coord')\nviz_df = pca_subset.join(lr_results,on=['id','features']).select('x_coord','y_coord','prediction','label')\nviz_df = viz_df.dropna(how='any',subset=['x_coord','y_coord']).toPandas()\n\na = sns.lmplot(x='x_coord',y='y_coord',hue='prediction',col='label',data=viz_df,\n           fit_reg=False,palette={1:'red',0:'green'},size=6,aspect=1)\na.set_titles(col_template=['Decision Tree results when label==\"Benign\"','Decision Tree results when label==\"Malignant\"'])", 
            "outputs": [], 
            "execution_count": 11
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "#### Classification metrics ####\ndef calc_metrics(results):\n    metrics = {}\n    metrics['tp_0'] = results.filter((results.label==0)&(results.prediction==0)).count()\n    metrics['fn_0'] = results.filter((results.label==0)&(results.prediction==1)).count()\n    metrics['tn_0'] = results.filter((results.label==1)&(results.prediction==1)).count()\n    metrics['fp_0'] = results.filter((results.label==1)&(results.prediction==0)).count()\n    \n    metrics['tp_1'] = results.filter((results.label==1)&(results.prediction==1)).count()\n    metrics['fn_1'] = results.filter((results.label==1)&(results.prediction==0)).count()\n    metrics['tn_1'] = results.filter((results.label==0)&(results.prediction==0)).count()\n    metrics['fp_1'] = results.filter((results.label==0)&(results.prediction==1)).count()\n    \n    return metrics\n        \n### calc precision & recall ###\ndef precision_recall(results):\n    items = calc_metrics(results)\n    rows = []\n    \n    row = {'class':0,'precision':items['tp_0']/float((items['tp_0']+items['fp_0'])+1),'recall':items['tp_0']/float((items['tp_0']+items['fn_0'])+1)}\n    rows.append(row)\n    row = {'class':1,'precision':items['tp_1']/float((items['tp_1']+items['fp_1'])+1),'recall':items['tp_1']/float((items['tp_1']+items['fn_1'])+1)}\n    rows.append(row)\n    \n    return rows", 
            "outputs": [], 
            "execution_count": 12
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Decision Tree metrics "
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "# pd.DataFrame(precision_recall(dt_results))", 
            "outputs": [], 
            "execution_count": 13
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "#### Logistic regression metrics "
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "# pd.DataFrame(precision_recall(lr_results))", 
            "outputs": [], 
            "execution_count": 14
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {}, 
            "source": "### Clustering Pipeline\n- K-means clustering"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "scrolled": true
            }, 
            "source": "######## Kmeans clustering, join to the PCA set #########\nfrom numpy import array\nfrom math import sqrt\nfrom pyspark.ml.clustering import KMeans\n\nkmeans = KMeans(k=2,featuresCol='features')\nmodel = kmeans.fit(pca_result)\ncenters = model.clusterCenters()\nkmeans_result = model.transform(pca_result).select('id','features','prediction')\n\nviz_df = pca_result.join(kmeans_result,on=['id','features'],how='inner')\\\n                   .select('id','features','x_coord','y_coord','prediction','label')\n\nviz_df = viz_df.dropna(how='any',subset=['x_coord','y_coord']).toPandas()\n\nsns.lmplot(x='x_coord',y='y_coord',hue='prediction',col='label',data=viz_df,\n           fit_reg=False,palette={1:'red',0:'green'},size=6,aspect=1)", 
            "outputs": [], 
            "execution_count": 8
        }, 
        {
            "cell_type": "markdown", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "### Deploy to Cloud by using Watson Machine Learning Repo\n   Please follow demonstration! :D\n"
        }, 
        {
            "cell_type": "code", 
            "metadata": {
                "collapsed": true
            }, 
            "source": "", 
            "outputs": [], 
            "execution_count": null
        }
    ], 
    "nbformat": 4, 
    "metadata": {
        "kernelspec": {
            "display_name": "Python 2 with Spark 2.0", 
            "name": "python2-spark20", 
            "language": "python"
        }, 
        "language_info": {
            "version": "2.7.11", 
            "nbconvert_exporter": "python", 
            "mimetype": "text/x-python", 
            "codemirror_mode": {
                "version": 2, 
                "name": "ipython"
            }, 
            "file_extension": ".py", 
            "name": "python", 
            "pygments_lexer": "ipython2"
        }
    }, 
    "nbformat_minor": 1
}