Lab 5 - Scalable PCA for dimensionality reduction.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# COM6012 Scalable Machine Learning 2019 - Haiping Lu\n",
    "# Lab 5: PCA for dimensionality reduction\n",
    "\n",
    "## Objectives\n",
    "\n",
    "* Task 1: To finish before the next session. **Essential**\n",
    "* Task 2: To finish before the next session. **Essential**\n",
    "* Task 3: To finish before the next session. **Essential**\n",
    "* Task 4: To explore by yourself. **Optional but recommended**\n",
    "\n",
    "**Suggested reading**: \n",
    "* [Extracting, transforming and selecting features](https://spark.apache.org/docs/2.3.2/ml-features.html)\n",
    "* [PCA in Spark](https://spark.apache.org/docs/2.3.2/ml-features.html#pca)\n",
    "* [SVD in Spark: RDD-based only `pyspark.mllib`](https://spark.apache.org/docs/2.3.2/mllib-dimensionality-reduction.html#singular-value-decomposition-svd), **not available** yet in DataFrame-based `pyspark.ml`\n",
    "* [StandardScaler in Spark](https://spark.apache.org/docs/2.3.2/ml-features.html#standardscaler) to standardise/normalise data to unit standard deviation and/or zero mean.\n",
    "* [Data Types - RDD-based API](https://spark.apache.org/docs/2.3.2/mllib-data-types.html)\n",
    "* [PCA on Wiki](https://en.wikipedia.org/wiki/Principal_component_analysis)\n",
    "* [Understanding Dimension Reduction with Principal Component Analysis (PCA)](https://blog.paperspace.com/dimension-reduction-with-principal-component-analysis/)\n",
    "* [Principal Component Analysis explained on Kaggle](https://www.kaggle.com/nirajvermafcb/principal-component-analysis-explained) with data available [here](https://www.kaggle.com/liujiaqi/hr-comma-sepcsv), and background info [here](https://rstudio-pubs-static.s3.amazonaws.com/345463_37f54d1c948b4cdfa181541841e0db8a.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://github.com/haipinglu/ScalableML/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import findspark\n",
    "#findspark.init()\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"COM6012 PCA\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Data Types: Local vs Distributed, Dense vs Sparse, Labelled vs Unlabelled\n",
    "\n",
    "To deal with data efficiently, Spark considers different [data types](https://spark.apache.org/docs/2.3.2/mllib-data-types.html).\n",
    "\n",
    "### Local vector:  Dense vs Sparse\n",
    "> A local vector has integer-typed and 0-based indices and double-typed values, stored on a single machine. MLlib supports two types of local vectors: dense and sparse. A dense vector is backed by a double array representing its entry values, while a sparse vector is backed by two parallel arrays: indices and values. For example, a vector (1.0, 0.0, 3.0) can be represented in dense format as [1.0, 0.0, 3.0] or in sparse format as (3, [0, 2], [1.0, 3.0]), where 3 is the size of the vector.\n",
    "\n",
    "Please check out the [Vector in RDD API](https://spark.apache.org/docs/2.3.2/api/python/pyspark.mllib.html#pyspark.mllib.linalg.Vectors) or [Vector in DataFrame API](https://spark.apache.org/docs/2.3.2/api/python/pyspark.ml.html#pyspark.ml.linalg.Vector)(see method `.Sparse()`) and [SparseVector in RDD API ](https://spark.apache.org/docs/2.3.2/api/python/pyspark.mllib.html#pyspark.mllib.linalg.SparseVector) or [SparseVector in DataFrame API ](https://spark.apache.org/docs/2.3.2/api/python/pyspark.ml.html#pyspark.ml.linalg.SparseVector). The official example is below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import scipy.sparse as sps\n",
    "from pyspark.mllib.linalg import Vectors\n",
    "\n",
    "# Use a NumPy array as a dense vector.\n",
    "dv1 = np.array([1.0, 0.0, 3.0])\n",
    "# Use a Python list as a dense vector.\n",
    "dv2 = [1.0, 0.0, 3.0]\n",
    "# Create a SparseVector.\n",
    "sv1 = Vectors.sparse(3, [0, 2], [1.0, 3.0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the vector created by `Vectors.sparse()` is of type `SparseVector()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "SparseVector(3, {0: 1.0, 2: 3.0})"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sv1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "View the sparse vector in a dense format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([1., 0., 3.])"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sv1.toArray()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Labeled point\n",
    "> A labeled point is a local vector, either dense or sparse, associated with a label/response. In MLlib, labeled points are used in supervised learning algorithms. We use a double to store a label, so we can use labeled points in both regression and classification. For binary classification, a label should be either 0 (negative) or 1 (positive). For multiclass classification, labels should be class indices starting from zero: 0, 1, 2, ....\n",
    "\n",
    "[LabeledPoint API](https://spark.apache.org/docs/2.3.2/api/python/pyspark.mllib.html#pyspark.mllib.regression.LabeledPoint)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import SparseVector\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "\n",
    "# Create a labeled point with a positive label and a dense feature vector.\n",
    "pos = LabeledPoint(1.0, [1.0, 0.0, 3.0])\n",
    "\n",
    "# Create a labeled point with a negative label and a sparse feature vector.\n",
    "neg = LabeledPoint(0.0, SparseVector(3, [0, 2], [1.0, 3.0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0.0,(3,[0,2],[1.0,3.0]))\n"
     ]
    }
   ],
   "source": [
    "print(neg)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "neg.label"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "SparseVector(3, {0: 1.0, 2: 3.0})"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "neg.features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "View the features as dense vector (rather than sparse vector)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([1., 0., 3.])"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "neg.features.toArray()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Local matrix\n",
    "> A local matrix has integer-typed row and column indices and double-typed values, stored on a single machine. MLlib supports dense matrices, whose entry values are stored in a single double array in column-major order, and sparse matrices, whose non-zero entry values are stored in the Compressed Sparse Column (CSC) format in column-major order. For example, the following dense matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import Matrix, Matrices\n",
    "\n",
    "# Create a dense matrix ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))\n",
    "dm2 = Matrices.dense(3, 2, [1, 3, 5, 2, 4, 6]) #Note the official example is wrong\n",
    "\n",
    "# Create a sparse matrix ((9.0, 0.0), (0.0, 8.0), (0.0, 6.0))\n",
    "sm = Matrices.sparse(3, 2, [0, 1, 3], [0, 2, 1], [9, 6, 8])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[1., 2.],\n",
      "             [3., 4.],\n",
      "             [5., 6.]])\n"
     ]
    }
   ],
   "source": [
    "print(dm2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3 X 2 CSCMatrix\n",
      "(0,0) 9.0\n",
      "(2,1) 6.0\n",
      "(1,1) 8.0\n"
     ]
    }
   ],
   "source": [
    "print(sm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See the [Compressed sparse column (CSC or CCS) format](https://en.wikipedia.org/wiki/Sparse_matrix#Compressed_sparse_column_(CSC_or_CCS))\n",
    "> values are read first by column, a row index is stored for each value, and column pointers are stored. For example, CSC is (val, row_ind, col_ptr), where val is an array of the (top-to-bottom, then left-to-right) non-zero values of the matrix; row_ind is the row indices corresponding to the values; and, col_ptr is the list of val indexes where each column starts. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[9., 0.],\n",
      "             [0., 8.],\n",
      "             [0., 6.]])\n"
     ]
    }
   ],
   "source": [
    "dsm=sm.toDense()\n",
    "print(dsm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distributed matrix\n",
    "\n",
    "> A distributed matrix has long-typed row and column indices and double-typed values, stored distributively in one or more RDDs. It is very important to choose the right format to store large and distributed matrices. Converting a distributed matrix to a different format may require a global shuffle, which is quite expensive. Four types of distributed matrices have been implemented so far.\n",
    "\n",
    "#### RowMatrix\n",
    "> The basic type is called RowMatrix. A RowMatrix is a row-oriented distributed matrix without meaningful row indices, e.g., a collection of feature vectors. It is backed by an RDD of its rows, where each row is a local vector. We assume that the number of columns is not huge for a RowMatrix so that a single local vector can be reasonably communicated to the driver and can also be stored / operated on using a single node.\n",
    "> Since each row is represented by a local vector, the number of columns is limited by the integer range but it should be much smaller in practice."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg.distributed import RowMatrix\n",
    "\n",
    "# Create an RDD of vectors.\n",
    "rows = sc.parallelize([[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12]])\n",
    "\n",
    "# Create a RowMatrix from an RDD of vectors.\n",
    "mat = RowMatrix(rows)\n",
    "\n",
    "# Get its size.\n",
    "m = mat.numRows()  # 4\n",
    "n = mat.numCols()  # 3\n",
    "\n",
    "# Get the rows as an RDD of vectors again.\n",
    "rowsRDD = mat.rows"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "View the RowMatrix in a dense matrix format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DenseVector([1.0, 2.0, 3.0]),\n",
       " DenseVector([4.0, 5.0, 6.0]),\n",
       " DenseVector([7.0, 8.0, 9.0]),\n",
       " DenseVector([10.0, 11.0, 12.0])]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rowsRDD.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. PCA\n",
    "\n",
    "[Principal component analysis](http://en.wikipedia.org/wiki/Principal_component_analysis) (PCA) is a statistical procedure that uses an orthogonal transformation to convert a set of observations of possibly correlated variables (entities each of which takes on various numerical values) into a set of values of linearly uncorrelated variables called **principal components (PCs)**. A PCA class trains a model to project vectors to a low-dimensional space using PCA and this is probably the most commonly used **dimensionality reduction** method.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PCA in DataFrame-based API `pyspark.ml`  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check out the [API](https://spark.apache.org/docs/2.3.2/api/python/pyspark.ml.html#pyspark.ml.feature.PCA). Check `pyspark.ml.feature.PCAModel` too to see what is available for the fitted model. The official example of projecting 5-dimensional feature vectors into 3-dimensional principal components."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/haiping/anaconda3/lib/python3.6/importlib/_bootstrap.py:219: RuntimeWarning: numpy.dtype size changed, may indicate binary incompatibility. Expected 96, got 88\n",
      "  return f(*args, **kwds)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+\n",
      "|pcaFeatures                                                |\n",
      "+-----------------------------------------------------------+\n",
      "|[1.6485728230883807,-4.013282700516296,-5.524543751369388] |\n",
      "|[-4.645104331781534,-1.1167972663619026,-5.524543751369387]|\n",
      "|[-6.428880535676489,-5.337951427775355,-5.524543751369389] |\n",
      "+-----------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import PCA\n",
    "from pyspark.ml.linalg import Vectors\n",
    "\n",
    "data = [(Vectors.sparse(5, [(1, 1.0), (3, 7.0)]),),\n",
    "        (Vectors.dense([2.0, 0.0, 3.0, 4.0, 5.0]),),\n",
    "        (Vectors.dense([4.0, 0.0, 0.0, 6.0, 7.0]),)]\n",
    "df = spark.createDataFrame(data, [\"features\"])\n",
    "\n",
    "pca = PCA(k=3, inputCol=\"features\", outputCol=\"pcaFeatures\")\n",
    "model = pca.fit(df)\n",
    "\n",
    "result = model.transform(df).select(\"pcaFeatures\")\n",
    "result.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|            features|\n",
      "+--------------------+\n",
      "| (5,[1,3],[1.0,7.0])|\n",
      "|[2.0,0.0,3.0,4.0,...|\n",
      "|[4.0,0.0,0.0,6.0,...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check the explained variance in percentage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([0.7944, 0.2056, 0.0])"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model.explainedVariance"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The variance should sum to one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9999999999999999"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum(model.explainedVariance)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Take a look at the principal components Matrix. Each column is one principal component."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The **PC** obtained by this method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[-0.44859172, -0.28423808,  0.08344545],\n",
      "             [ 0.13301986, -0.05621156,  0.04423979],\n",
      "             [-0.12523156,  0.76362648, -0.57807123],\n",
      "             [ 0.21650757, -0.56529588, -0.79554051],\n",
      "             [-0.84765129, -0.11560341, -0.15501179]])\n"
     ]
    }
   ],
   "source": [
    "print(model.pc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PCA in RDD-based API `pyspark.mllib` "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Eigendecomposition for PCA\n",
    "\n",
    "`pyspark.mllib` supports PCA for **tall-and-skinny** (big $n$, small $d$) matrices stored in row-oriented format and any Vectors. We demonstrate how to compute principal components on a [<tt>RowMatrix</tt>](http://spark.apache.org/docs/latest/mllib-data-types.html#rowmatrix) and use them to project the vectors into a low-dimensional space in the cell below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import Vectors\n",
    "from pyspark.mllib.linalg.distributed import RowMatrix\n",
    "\n",
    "rows = sc.parallelize([\n",
    "    Vectors.sparse(5, {1: 1.0, 3: 7.0}),\n",
    "    Vectors.dense(2.0, 0.0, 3.0, 4.0, 5.0),\n",
    "    Vectors.dense(4.0, 0.0, 0.0, 6.0, 7.0)\n",
    "])\n",
    "\n",
    "mat = RowMatrix(rows)\n",
    "# Compute the top 3 principal components.\n",
    "# Principal components are stored in a local dense matrix.\n",
    "pc = mat.computePrincipalComponents(3)\n",
    "\n",
    "# Project the rows to the linear space spanned by the top 4 principal components.\n",
    "projected = mat.multiply(pc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[SparseVector(5, {1: 1.0, 3: 7.0}),\n",
       " DenseVector([2.0, 0.0, 3.0, 4.0, 5.0]),\n",
       " DenseVector([4.0, 0.0, 0.0, 6.0, 7.0])]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rows.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we convert to dense rows to see the matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DenseVector([0.0, 1.0, 0.0, 7.0, 0.0]),\n",
       " DenseVector([2.0, 0.0, 3.0, 4.0, 5.0]),\n",
       " DenseVector([4.0, 0.0, 0.0, 6.0, 7.0])]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.mllib.linalg import DenseVector\n",
    "\n",
    "denseRows = rows.map(lambda vector: DenseVector(vector.toArray()))\n",
    "denseRows.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The **PC** obtained by this method (the same as the above)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[-0.44859172, -0.28423808,  0.08344545],\n",
      "             [ 0.13301986, -0.05621156,  0.04423979],\n",
      "             [-0.12523156,  0.76362648, -0.57807123],\n",
      "             [ 0.21650757, -0.56529588, -0.79554051],\n",
      "             [-0.84765129, -0.11560341, -0.15501179]])\n"
     ]
    }
   ],
   "source": [
    "print(pc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DenseVector([1.6486, -4.0133, -5.5245]),\n",
       " DenseVector([-4.6451, -1.1168, -5.5245]),\n",
       " DenseVector([-6.4289, -5.338, -5.5245])]"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "projected.rows.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The same as above (see `pcaFeatures` output)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### SVD for PCA  - more **scalable** way to do PCA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please read [SVD in RDD-based API `pyspark.mllib`](https://spark.apache.org/docs/2.3.2/mllib-dimensionality-reduction.html#singular-value-decomposition-svd). As covered in the lecture, we will need SVD for PCA on large-scale data. Here, we use it on the same small toy example to examine the relationship with eigenvalue decomposition based PCA methods above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute the top 5 singular values and corresponding singular vectors.\n",
    "svd = mat.computeSVD(3, computeU=True)\n",
    "U = svd.U       # The U factor is a RowMatrix.\n",
    "s = svd.s       # The singular values are stored in a local dense vector.\n",
    "V = svd.V       # The V factor is a local dense matrix."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we are doing it right, the right singular vectors should be the same as the eigenvectors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[-0.31278534,  0.31167136,  0.30366911],\n",
      "             [-0.02980145, -0.17133211, -0.02226069],\n",
      "             [-0.12207248,  0.15256471, -0.95070998],\n",
      "             [-0.71847899, -0.68096285, -0.0172245 ],\n",
      "             [-0.60841059,  0.62170723,  0.05606596]])\n"
     ]
    }
   ],
   "source": [
    "print(V)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But it is **not the same**! Why? We need to do **centering**! We can do so use the [StandardScaler (check out the API](https://spark.apache.org/docs/2.3.2/mllib-feature-extraction.html#standardscaler))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.feature import StandardScaler\n",
    "\n",
    "#We center the data to remove the mean. \n",
    "standardizer = StandardScaler(True, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DenseVector([-2.0, 0.6667, -1.0, 1.3333, -4.0]),\n",
       " DenseVector([0.0, -0.3333, 2.0, -1.6667, 1.0]),\n",
       " DenseVector([2.0, -0.3333, -1.0, 0.3333, 3.0])]"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model = standardizer.fit(rows)\n",
    "centeredRows = model.transform(rows)\n",
    "centeredRows.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "centeredmat = RowMatrix(centeredRows)\n",
    "\n",
    "# Compute the top 3 singular values and corresponding singular vectors.\n",
    "svd = centeredmat.computeSVD(3, computeU=True)\n",
    "U = svd.U       # The U factor is a RowMatrix.\n",
    "s = svd.s       # The singular values are stored in a local dense vector.\n",
    "V = svd.V       # The V factor is a local dense matrix."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check the **PC** obtained this time (it is the same as the above PCA methods now)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DenseMatrix([[-0.44859172, -0.28423808, -0.81664677],\n",
      "             [ 0.13301986, -0.05621156, -0.03622012],\n",
      "             [-0.12523156,  0.76362648, -0.34267361],\n",
      "             [ 0.21650757, -0.56529588, -0.13898906],\n",
      "             [-0.84765129, -0.11560341,  0.44162541]])\n"
     ]
    }
   ],
   "source": [
    "print(V)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[6.001041088520536,3.0530049438580336,5.017337590814169e-08]\n"
     ]
    }
   ],
   "source": [
    "print(s)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We get the eigenvalues by taking squares of the singular values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([36.0125, 9.3208, 0.0])"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evs=s*s\n",
    "evs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we compute the percentage of variance captures and compare with the above to verify (see/search `model.explainedVariance`)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([0.7944, 0.2056, 0.0])"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evs/sum(evs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Exercise - Further analysis of the Iris data (completing three or more questions is considered as completion of this exercise).\n",
    "\n",
    "* The [HR Analytics problem:](https://rstudio-pubs-static.s3.amazonaws.com/345463_37f54d1c948b4cdfa181541841e0db8a.html)  A company is trying to figure out why their best and experienced employees are leaving prematurely from a [dataset](https://www.kaggle.com/liujiaqi/hr-comma-sepcsv). Follow the example [Principal Component Analysis explained on Kaggle](https://www.kaggle.com/nirajvermafcb/principal-component-analysis-explained) to perform such analysis in PySpark, using as many PySpark APIs as possible.\n",
    "* Follow [Understanding Dimension Reduction with Principal Component Analysis (PCA)](https://blog.paperspace.com/dimension-reduction-with-principal-component-analysis/) to use PCA in inspecting the Iris data in Lab 4.\n",
    "* Use `pca.fit()`, `computePrincipalComponents()` and `computeSVD` to separately compute the PCs of the same data and compare/verify whether you obtain the same results.\n",
    "* Choose a bigger dataset and use the three ways above to separately compute the PCs and compare the time taken. \n",
    "* Choose datasets of increasing sizes and use the three ways above to compute the PCs and plot the time cost against the size to study the scalability\n",
    "* Increase the number of cores used, e.g., setting `local[4]`, `local[6]`, `local[8]`, and study the time cost against the number of cores used.\n",
    "\n",
    "You can find datasets of your own interest, e.g., face/digit datasets explored in MLAI. You may also use consider using [Bag of Words Data Sets](https://archive.ics.uci.edu/ml/datasets/Bag+of+Words) below in Task 3, which have data sets of various sizes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. PCA on bigger data (Optional but recommended)\n",
    "\n",
    "### Word meaning extraction\n",
    "\n",
    "Use PySpark to perform the steps in IBM's notebook on [Spark-based machine learning for word meanings](https://dataplatform.cloud.ibm.com/analytics/notebooks/cd12e0c6-3560-4ad2-9850-82055daada23/view?access_token=4b8b2a7ac1f15b80e93aa12b93b60344368bea0014f2c48a7c6bc45a06a90ef3) that makes use of PCA, kmeans, and Word2Vec to learn work meanings.\n",
    "\n",
    "### Bag of words analysis\n",
    "\n",
    "Choose a [Bag of Words Data Set](https://archive.ics.uci.edu/ml/datasets/Bag+of+Words). Let us take  the **NIPS full papers** data as an example. \n",
    "\n",
    "The format of this data is\n",
    "\n",
    "    Number of documents\n",
    "    Number of words in the vocabulary\n",
    "    Total number of words in the collection\n",
    "    docID wordID count\n",
    "    docID wordID count\n",
    "    ...\n",
    "    docID wordID count\n",
    "    \n",
    "Our data matrix will be $doc \\times wordcount$ Initially, we need to read this data in: the steps in this would be roughly:\n",
    "\n",
    "1. extract the number of documents, size of the vocabulary and strip off the first 3 lines\n",
    "2. combine the words per document\n",
    "3. create sparse vectors (for better space efficiency)\n",
    "\n",
    "Create this as a standalone program. Keep everything as parallel as possible. You will benefit from creating a very small example to test your work, and then checking **whether** your work scales up to the **NYTIMES** bagofwords data.\n",
    "\n",
    "### Large image datasets\n",
    "\n",
    "Find some large-scale image datasets to examine the principal components and explore low-dimensional representations. "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}