CS246_Colab_3.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "CS246_Colab_3.ipynb",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "accelerator": "TPU"
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kPt5q27L5557"
      },
      "source": [
        "# CS246 - Colab 3\n",
        "## K-Means & PCA"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "p0-YhEpP_Ds-"
      },
      "source": [
        "### Setup"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zsj5WYpR9QId"
      },
      "source": [
        "Let's setup Spark on your Colab environment.  Run the cell below!"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k-qHai2252mI",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "ade90b34-32d0-42d8-9a2b-bbb27bf9b17e"
      },
      "source": [
        "!pip install pyspark\n",
        "!pip install -U -q PyDrive\n",
        "!apt install openjdk-8-jdk-headless -qq\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\""
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.1.2.tar.gz (212.4 MB)\n",
            "\u001b[K     |████████████████████████████████| 212.4 MB 67 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9\n",
            "  Downloading py4j-0.10.9-py2.py3-none-any.whl (198 kB)\n",
            "\u001b[K     |████████████████████████████████| 198 kB 47.0 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.1.2-py2.py3-none-any.whl size=212880768 sha256=9c8543838a67c70a309465df40c06a1ec917adf697f58d31a5a7dd254755bd67\n",
            "  Stored in directory: /root/.cache/pip/wheels/a5/0a/c1/9561f6fecb759579a7d863dcd846daaa95f598744e71b02c77\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9 pyspark-3.1.2\n",
            "The following additional packages will be installed:\n",
            "  openjdk-8-jre-headless\n",
            "Suggested packages:\n",
            "  openjdk-8-demo openjdk-8-source libnss-mdns fonts-dejavu-extra\n",
            "  fonts-ipafont-gothic fonts-ipafont-mincho fonts-wqy-microhei\n",
            "  fonts-wqy-zenhei fonts-indic\n",
            "The following NEW packages will be installed:\n",
            "  openjdk-8-jdk-headless openjdk-8-jre-headless\n",
            "0 upgraded, 2 newly installed, 0 to remove and 40 not upgraded.\n",
            "Need to get 36.5 MB of archives.\n",
            "After this operation, 143 MB of additional disk space will be used.\n",
            "Selecting previously unselected package openjdk-8-jre-headless:amd64.\n",
            "(Reading database ... 160837 files and directories currently installed.)\n",
            "Preparing to unpack .../openjdk-8-jre-headless_8u292-b10-0ubuntu1~18.04_amd64.deb ...\n",
            "Unpacking openjdk-8-jre-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "Selecting previously unselected package openjdk-8-jdk-headless:amd64.\n",
            "Preparing to unpack .../openjdk-8-jdk-headless_8u292-b10-0ubuntu1~18.04_amd64.deb ...\n",
            "Unpacking openjdk-8-jdk-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "Setting up openjdk-8-jre-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/orbd to provide /usr/bin/orbd (orbd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/servertool to provide /usr/bin/servertool (servertool) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/tnameserv to provide /usr/bin/tnameserv (tnameserv) in auto mode\n",
            "Setting up openjdk-8-jdk-headless:amd64 (8u292-b10-0ubuntu1~18.04) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/idlj to provide /usr/bin/idlj (idlj) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsimport to provide /usr/bin/wsimport (wsimport) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jsadebugd to provide /usr/bin/jsadebugd (jsadebugd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/native2ascii to provide /usr/bin/native2ascii (native2ascii) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/javah to provide /usr/bin/javah (javah) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/hsdb to provide /usr/bin/hsdb (hsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/clhsdb to provide /usr/bin/clhsdb (clhsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/extcheck to provide /usr/bin/extcheck (extcheck) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/schemagen to provide /usr/bin/schemagen (schemagen) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/xjc to provide /usr/bin/xjc (xjc) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jhat to provide /usr/bin/jhat (jhat) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsgen to provide /usr/bin/wsgen (wsgen) in auto mode\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qwtlO4_m_LbQ"
      },
      "source": [
        "Now we import some of the libraries usually needed by our workload.\n",
        "\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "twk-K-jilWK7"
      },
      "source": [
        "import pandas as pd\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "%matplotlib inline\n",
        "\n",
        "import pyspark\n",
        "from pyspark.sql import *\n",
        "from pyspark.sql.types import *\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark import SparkContext, SparkConf"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BtrJlMBt1Ela"
      },
      "source": [
        "Let's initialize the Spark context."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Vm3sAVeK1EDZ"
      },
      "source": [
        "# create the session\n",
        "conf = SparkConf().set(\"spark.ui.port\", \"4050\")\n",
        "\n",
        "# create the context\n",
        "sc = pyspark.SparkContext(conf=conf)\n",
        "spark = SparkSession.builder.getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YqovskkH1DmC"
      },
      "source": [
        "You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "DueQggJc1DDk",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 216
        },
        "outputId": "1caedd02-6987-4d14-bafe-3671d5768952"
      },
      "source": [
        "spark"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "\n",
              "            <div>\n",
              "                <p><b>SparkSession - in-memory</b></p>\n",
              "                \n",
              "        <div>\n",
              "            <p><b>SparkContext</b></p>\n",
              "\n",
              "            <p><a href=\"http://6a028bbdff61:4050\">Spark UI</a></p>\n",
              "\n",
              "            <dl>\n",
              "              <dt>Version</dt>\n",
              "                <dd><code>v3.1.2</code></dd>\n",
              "              <dt>Master</dt>\n",
              "                <dd><code>local[*]</code></dd>\n",
              "              <dt>AppName</dt>\n",
              "                <dd><code>pyspark-shell</code></dd>\n",
              "            </dl>\n",
              "        </div>\n",
              "        \n",
              "            </div>\n",
              "        "
            ],
            "text/plain": [
              "<pyspark.sql.session.SparkSession at 0x7faecfade550>"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 4
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Iid7lXcG1CY8"
      },
      "source": [
        "If you are running this Colab on the Google hosted runtime, the cell below will create a *ngrok* tunnel which will allow you to still check the Spark UI."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sDnGLVPH1BPQ",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "157d9ebb-228e-4e65-e652-b3b0e9ef5935"
      },
      "source": [
        "!wget https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip\n",
        "!unzip ngrok-stable-linux-amd64.zip\n",
        "get_ipython().system_raw('./ngrok http 4050 &')\n",
        "!curl -s http://localhost:4040/api/tunnels | python3 -c \\\n",
        "    \"import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])\""
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "--2021-08-09 11:57:18--  https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip\n",
            "Resolving bin.equinox.io (bin.equinox.io)... 35.171.130.195, 3.226.157.132, 34.239.30.50, ...\n",
            "Connecting to bin.equinox.io (bin.equinox.io)|35.171.130.195|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 13832437 (13M) [application/octet-stream]\n",
            "Saving to: ‘ngrok-stable-linux-amd64.zip’\n",
            "\n",
            "ngrok-stable-linux- 100%[===================>]  13.19M  39.6MB/s    in 0.3s    \n",
            "\n",
            "2021-08-09 11:57:18 (39.6 MB/s) - ‘ngrok-stable-linux-amd64.zip’ saved [13832437/13832437]\n",
            "\n",
            "Archive:  ngrok-stable-linux-amd64.zip\n",
            "  inflating: ngrok                   \n",
            "Traceback (most recent call last):\n",
            "  File \"<string>\", line 1, in <module>\n",
            "IndexError: list index out of range\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kAYRX2PMm0L6"
      },
      "source": [
        "### Data Preprocessing"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7hXdMR6wnEIM"
      },
      "source": [
        "In this Colab, rather than downloading a file from Google Drive, we will load a famous machine learning dataset, the [Breast Cancer Wisconsin dataset](https://scikit-learn.org/stable/modules/generated/sklearn.datasets.load_breast_cancer.html), using the ```scikit-learn``` datasets loader."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "5K93ABEy9Zlo"
      },
      "source": [
        "from sklearn.datasets import load_breast_cancer\n",
        "breast_cancer = load_breast_cancer()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wpsaYOqRxar2"
      },
      "source": [
        "For convenience, given that the dataset is small, we first construct a Pandas dataframe, tune the schema, and then convert it into a Spark dataframe."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Oitav_xhQD9w",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f9b7be25-ef08-44ef-c0d2-b70e04719111"
      },
      "source": [
        "pd_df = pd.DataFrame(breast_cancer.data, columns=breast_cancer.feature_names)\n",
        "df = spark.createDataFrame(pd_df)\n",
        "\n",
        "def set_df_columns_nullable(spark, df, column_list, nullable=False):\n",
        "    for struct_field in df.schema:\n",
        "        if struct_field.name in column_list:\n",
        "            struct_field.nullable = nullable\n",
        "    df_mod = spark.createDataFrame(df.rdd, df.schema)\n",
        "    return df_mod\n",
        "\n",
        "df = set_df_columns_nullable(spark, df, df.columns)\n",
        "df = df.withColumn('features', array(df.columns))\n",
        "vectors = df.rdd.map(lambda row: Vectors.dense(row.features))\n",
        "\n",
        "df.printSchema()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- mean radius: double (nullable = false)\n",
            " |-- mean texture: double (nullable = false)\n",
            " |-- mean perimeter: double (nullable = false)\n",
            " |-- mean area: double (nullable = false)\n",
            " |-- mean smoothness: double (nullable = false)\n",
            " |-- mean compactness: double (nullable = false)\n",
            " |-- mean concavity: double (nullable = false)\n",
            " |-- mean concave points: double (nullable = false)\n",
            " |-- mean symmetry: double (nullable = false)\n",
            " |-- mean fractal dimension: double (nullable = false)\n",
            " |-- radius error: double (nullable = false)\n",
            " |-- texture error: double (nullable = false)\n",
            " |-- perimeter error: double (nullable = false)\n",
            " |-- area error: double (nullable = false)\n",
            " |-- smoothness error: double (nullable = false)\n",
            " |-- compactness error: double (nullable = false)\n",
            " |-- concavity error: double (nullable = false)\n",
            " |-- concave points error: double (nullable = false)\n",
            " |-- symmetry error: double (nullable = false)\n",
            " |-- fractal dimension error: double (nullable = false)\n",
            " |-- worst radius: double (nullable = false)\n",
            " |-- worst texture: double (nullable = false)\n",
            " |-- worst perimeter: double (nullable = false)\n",
            " |-- worst area: double (nullable = false)\n",
            " |-- worst smoothness: double (nullable = false)\n",
            " |-- worst compactness: double (nullable = false)\n",
            " |-- worst concavity: double (nullable = false)\n",
            " |-- worst concave points: double (nullable = false)\n",
            " |-- worst symmetry: double (nullable = false)\n",
            " |-- worst fractal dimension: double (nullable = false)\n",
            " |-- features: array (nullable = false)\n",
            " |    |-- element: double (containsNull = false)\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TtR1xRvonxiO"
      },
      "source": [
        "With the next cell, we build the two datastructures that we will be using throughout this Colab:\n",
        "\n",
        "\n",
        "*   ```features```, a dataframe of Dense vectors, containing all the original features in the dataset;\n",
        "*   ```labels```, a series of binary labels indicating if the corresponding set of features belongs to a subject with breast cancer, or not.\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "GP23Xkgwi0SD"
      },
      "source": [
        "from pyspark.ml.linalg import Vectors\n",
        "features = spark.createDataFrame(vectors.map(Row), [\"features\"])\n",
        "labels = pd.Series(breast_cancer.target)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CRaF2A_j_nC7"
      },
      "source": [
        "### Your task"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ebLNUxP0_8x3"
      },
      "source": [
        "If you run successfully the Setup and Data Preprocessing stages, you are now ready to cluster the data with the [K-means](https://spark.apache.org/docs/latest/ml-clustering.html) algorithm included in MLlib (Spark's Machine Learning library).\n",
        "Set the ```k``` parameter to **2**, fit the model, and the compute the [Silhouette score](https://en.wikipedia.org/wiki/Silhouette_(clustering)) (i.e., a measure of quality of the obtained clustering).  \n",
        "\n",
        "**IMPORTANT:** use the MLlib implementation of the Silhouette score (via ```ClusteringEvaluator```)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0xVIfPHZwWaE",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "0182ac5d-b712-4818-c42b-6f3379577f66"
      },
      "source": [
        "from pyspark.ml.clustering import KMeans\n",
        "from pyspark.ml.evaluation import ClusteringEvaluator\n",
        "\n",
        "kmeans = KMeans(k=2)\n",
        "model = kmeans.fit(df)\n",
        "predictions = model.transform(df)\n",
        "evaluator = ClusteringEvaluator()\n",
        "silhouette = evaluator.evaluate(predictions)\n",
        "\n",
        "print(\"Silhouette:\", silhouette)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Silhouette: 0.8342904262826145\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5GB09n7sqTO6"
      },
      "source": [
        "Take the predictions produced by K-means, and compare them with the ```labels``` variable (i.e., the ground truth from our dataset).  \n",
        "\n",
        "Compute how many data points in the dataset have been clustered correctly (i.e., positive cases in one cluster, negative cases in the other).\n",
        "\n",
        "*HINT*: you can use ```np.count_nonzero(series_a == series_b)``` to quickly compute the element-wise comparison of two series.\n",
        "\n",
        "**IMPORTANT**: K-means is a clustering algorithm, so it will not output a label for each data point, but just a cluster identifier!  As such, label ```0``` does not necessarily match the cluster identifier ```0```.\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WQhC3APIPPM5",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "59660f55-5bbd-4aa8-cdc1-ad1fbb52fa08"
      },
      "source": [
        "np.count_nonzero(np.array(predictions.select(\"prediction\").collect()).flatten() == labels)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "486"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 14
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fLIprM1JsdTU"
      },
      "source": [
        "Now perform dimensionality reduction on the ```features``` using the [PCA](https://spark.apache.org/docs/latest/ml-features.html#pca) statistical procedure, available as well in MLlib.\n",
        "\n",
        "Set the ```k``` parameter to **2**, effectively reducing the dataset size of a **15X** factor."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "p4J8JMDkSb24"
      },
      "source": [
        "from pyspark.ml.feature import PCA\n",
        "\n",
        "pca = PCA(k=2)\n",
        "model = pca.fit(df)\n",
        "transformed = model.transform(df)\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8leQR4-atMAl"
      },
      "source": [
        "Now run K-means with the same parameters as above, but on the ```pcaFeatures``` produced by the PCA reduction you just executed.\n",
        "\n",
        "Compute the Silhouette score, as well as the number of data points that have been clustered correctly."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "U_snSSj5k2y5"
      },
      "source": [
        "# YOUR CODE HERE"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sMUb_afmlIEW"
      },
      "source": [
        "# YOUR CODE HERE"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "SIrXJyVNP2AI"
      },
      "source": [
        "Once you obtained the desired results, **head over to Gradescope and submit your solution for this Colab**!"
      ]
    }
  ]
}