4. Spark MLib (Basics).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MLib is Machine Learning Library for Spark\n",
    "1. Incorporates with Numpy in python\n",
    "2. It provied an Integrated data analysis Flow\n",
    "3. Enhances speed and performances\n",
    "4. Clustering, Feature Pattern Matching, Linear Algebra, Collaborative filtering, classification, regression"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Spark  sparse and dense vector\n",
    "Spark Mlib supports both sparse and dense vector\n",
    "\n",
    "1. Dense vector contain zero values\n",
    "2. Sparse vector don not contain zero values as in huge vectors where most of the values are zero it becomes very inefficient to store all zero values\n",
    "\n",
    "#### Labelled Point\n",
    "1. A labeled point is a local vector, either dense or sparse, associated with a label/response.\n",
    "2. In MLlib, labeled points are used in supervised learning algorithms for regression and classification both."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import SparseVector\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "\n",
    "dense_vector = LabeledPoint(1.0, [1.0, 0, 3.0, 0, 0, 7.0])\n",
    "sparse_vector = LabeledPoint(1.0, SparseVector(6, [ 0, 2, 5], [1.0, 3.0, 7.0]))\n",
    "#here both have same meaning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Basic Statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "sc = pyspark.SparkContext(appName=\"mlib-stats\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['67.0,1.0,4.0,160.0,286.0,0.0,2.0,108.0,1.0,1.5,2.0,3.0,3.0,2',\n",
       " '67.0,1.0,4.0,120.0,229.0,0.0,2.0,129.0,1.0,2.6,2.0,2.0,7.0,1',\n",
       " '37.0,1.0,3.0,130.0,250.0,0.0,0.0,187.0,0.0,3.5,3.0,0.0,3.0,0',\n",
       " '41.0,0.0,2.0,130.0,204.0,0.0,2.0,172.0,0.0,1.4,1.0,0.0,3.0,0',\n",
       " '56.0,1.0,2.0,120.0,236.0,0.0,0.0,178.0,0.0,0.8,1.0,0.0,3.0,0']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "textRdd = sc.textFile(\"Resources/heart.csv\").filter(lambda x : x[0]!='a')\n",
    "textRdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def buildVectors(entry):\n",
    "    vector = entry.split(',')\n",
    "    return vector\n",
    "vectorRdd = textRdd.map(buildVectors)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[67.0, 1.0, 4.0, 160.0, 286.0, 0.0, 2.0, 108.0, 1.0, 1.5, 2.0, 3.0, 3.0, 2.0],\n",
       " [67.0, 1.0, 4.0, 120.0, 229.0, 0.0, 2.0, 129.0, 1.0, 2.6, 2.0, 2.0, 7.0, 1.0],\n",
       " [37.0, 1.0, 3.0, 130.0, 250.0, 0.0, 0.0, 187.0, 0.0, 3.5, 3.0, 0.0, 3.0, 0.0],\n",
       " [41.0, 0.0, 2.0, 130.0, 204.0, 0.0, 2.0, 172.0, 0.0, 1.4, 1.0, 0.0, 3.0, 0.0],\n",
       " [56.0, 1.0, 2.0, 120.0, 236.0, 0.0, 0.0, 178.0, 0.0, 0.8, 1.0, 0.0, 3.0, 0.0]]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def convertFloat(entry):\n",
    "    vector = []\n",
    "    for x in entry:\n",
    "        if x!='' and x!='?':\n",
    "            vector.append(float(x))\n",
    "        else:\n",
    "            vector.append('?')\n",
    "    return vector\n",
    "fvectorRdd = vectorRdd.map(convertFloat).filter(lambda x: '?' not in x)\n",
    "fvectorRdd.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Statistics\n",
    "colStats() returns an instance of MultivariateStatisticalSummary, which contains the column-wise max, min, mean, variance, and number of nonzeros, as well as the total count."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[  5.45135135e+01   6.75675676e-01   3.16554054e+00   1.31648649e+02\n",
      "   2.47398649e+02   1.41891892e-01   9.93243243e-01   1.49597973e+02\n",
      "   3.27702703e-01   1.05135135e+00   1.59797297e+00   6.79054054e-01\n",
      "   4.72635135e+00   9.49324324e-01]\n",
      "[  8.19320202e+01   2.19880898e-01   9.18266148e-01   3.15984608e+02\n",
      "   2.71221342e+03   1.22171324e-01   9.89784700e-01   5.28098843e+02\n",
      "   2.21060467e-01   1.35918461e+00   3.76809437e-01   8.83085204e-01\n",
      "   3.76554054e+00   1.52623683e+00]\n",
      "[ 296.  200.  296.  296.  296.   42.  149.  296.   97.  200.  296.  123.\n",
      "  296.  137.]\n"
     ]
    }
   ],
   "source": [
    "from pyspark.mllib.stat import Statistics\n",
    "summary = Statistics.colStats(fvectorRdd)\n",
    "print(summary.mean())\n",
    "print(summary.variance())\n",
    "print(summary.numNonzeros())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Correlation\n",
    "Statistics provides methods to calculate correlations between series."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.1323795611721055\n",
      "0.1406576227867661\n"
     ]
    }
   ],
   "source": [
    "x = fvectorRdd.map(lambda x : x[3])\n",
    "y = fvectorRdd.map(lambda x : x[4])\n",
    "print(Statistics.corr(x,y,method='pearson'))\n",
    "print(Statistics.corr(x,y,method=\"spearman\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}