spark-intro-ml-scala.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is implemented using Apache Toree\n",
    "\n",
    "When launching jupyter please add:\n",
    "\n",
    "  jupyter install --spark_opts=\"--packages com.databricks:spark-csv_2.10:1.3.0\"\n",
    "\n",
    "To your Toree setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "org.apache.spark.SparkContext@43c33e98"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we start by downloading loading some data which is in csv format so its a good thing we got that csv package included already for us.\n",
    "\n",
    "Note: the data is a modified version of https://archive.ics.uci.edu/ml/datasets/Adult"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val df = sqlContext.read.format(\"com.databricks.spark.csv\").option(\"header\", \"true\").load(\"resources/adult.data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: string (nullable = true)\n",
      " |-- workclass: string (nullable = true)\n",
      " |-- fnlwgt: string (nullable = true)\n",
      " |-- education: string (nullable = true)\n",
      " |-- education-num: string (nullable = true)\n",
      " |-- maritial-status: string (nullable = true)\n",
      " |-- occupation: string (nullable = true)\n",
      " |-- relationship: string (nullable = true)\n",
      " |-- race: string (nullable = true)\n",
      " |-- sex: string (nullable = true)\n",
      " |-- capital-gain: string (nullable = true)\n",
      " |-- capital-loss: string (nullable = true)\n",
      " |-- hours-per-week: string (nullable = true)\n",
      " |-- native-country: string (nullable = true)\n",
      " |-- category: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So as we can see Spark has simply loaded all of the values as strings since we haven't specified another schema. We can instead ask it to infer the schema and also handle this extra space magic:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val df = sqlContext.read.format(\"com.databricks.spark.csv\").option(\"header\", \"true\").option(\"inferSchema\", \"true\").load(\"./resources/adult.data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[age: int, workclass: string, fnlwgt: double, education: string, education-num: double, maritial-status: string, occupation: string, relationship: string, race: string, sex: string, capital-gain: double, capital-loss: double, hours-per-week: double, native-country: string, category: string]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.cache()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[39, State-gov,77516.0, Bachelors,13.0, Never-married, Adm-clerical, Not-in-family, White, Male,2174.0,0.0,40.0, United-States, <=50K]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: integer (nullable = true)\n",
      " |-- workclass: string (nullable = true)\n",
      " |-- fnlwgt: double (nullable = true)\n",
      " |-- education: string (nullable = true)\n",
      " |-- education-num: double (nullable = true)\n",
      " |-- maritial-status: string (nullable = true)\n",
      " |-- occupation: string (nullable = true)\n",
      " |-- relationship: string (nullable = true)\n",
      " |-- race: string (nullable = true)\n",
      " |-- sex: string (nullable = true)\n",
      " |-- capital-gain: double (nullable = true)\n",
      " |-- capital-loss: double (nullable = true)\n",
      " |-- hours-per-week: double (nullable = true)\n",
      " |-- native-country: string (nullable = true)\n",
      " |-- category: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we import the Vector and DecisionTreeModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.mllib.linalg.Vectors\n",
    "import org.apache.spark.ml.feature.VectorAssembler\n",
    "import org.apache.spark.ml.feature.StringIndexer\n",
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}\n",
    "\n",
    "import org.apache.spark.mllib.linalg.Vectors\n",
    "import org.apache.spark.ml.feature.VectorAssembler\n",
    "import org.apache.spark.ml.feature.StringIndexer\n",
    "import org.apache.spark.ml.Pipeline\n",
    "import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first step is prepairing the features, here we are just choosing existing numeric features:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val assembler = new VectorAssembler().setInputCols(Array(\"age\", \"education-num\")).setOutputCol(\"features\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now the pipeline only works on doubles, so we need to take our category and turn it into a double. The StringIndexer will do this for us:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val indexer = new StringIndexer().setInputCol(\"category\").setOutputCol(\"category-index\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make a pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val pipeline = new Pipeline().setStages(Array(assembler, indexer))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We then need to \"fit\" our pipeline. This allows the StringIndexer to determine what strings will be assigned what indexes in the eventual transformation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val model = pipeline.fit(df)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val prepared = model.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[39, State-gov,77516.0, Bachelors,13.0, Never-married, Adm-clerical, Not-in-family, White, Male,2174.0,0.0,40.0, United-States, <=50K,[39.0,13.0],0.0]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "prepared.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val classifier  = new DecisionTreeClassifier().setFeaturesCol(\"features\").setLabelCol(\"category-index\") \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And now we fit on the prepared data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val classifier_model = classifier.fit(prepared)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val pipeline_and_model = new Pipeline().setStages(Array(assembler,indexer,classifier_model))\n",
    "val pipeline_model = pipeline_and_model.fit(df)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1.0,0.0]\n",
      "[1.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[1.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,1.0]\n",
      "[0.0,1.0]\n",
      "[1.0,1.0]\n",
      "[0.0,1.0]\n",
      "[0.0,1.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,1.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[1.0,1.0]\n"
     ]
    }
   ],
   "source": [
    "classifier_model.transform(prepared).select(\"prediction\", \"category-index\").take(20).foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1.0,0.0]\n",
      "[1.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[1.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,1.0]\n",
      "[0.0,1.0]\n",
      "[1.0,1.0]\n",
      "[0.0,1.0]\n",
      "[0.0,1.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,1.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[0.0,0.0]\n",
      "[1.0,1.0]\n"
     ]
    }
   ],
   "source": [
    "pipeline_model.transform(df).select(\"prediction\", \"category-index\").take(20).foreach(println)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(\" <=50K\", \" >50K\")"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.StringIndexerModel\n",
    "\n",
    "val labels= (pipeline_model.stages(1).asInstanceOf[StringIndexerModel]).labels\n",
    "labels"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[ >50K, <=50K]\n",
      "[ >50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ >50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, >50K]\n",
      "[ <=50K, >50K]\n",
      "[ >50K, >50K]\n",
      "[ <=50K, >50K]\n",
      "[ <=50K, >50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, >50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ <=50K, <=50K]\n",
      "[ >50K, >50K]\n"
     ]
    }
   ],
   "source": [
    "import org.apache.spark.ml.feature.IndexToString\n",
    "\n",
    "val inverter = new IndexToString().setInputCol(\"prediction\").setOutputCol(\"prediction-label\").setLabels(labels)\n",
    "\n",
    "inverter.transform(pipeline_model.transform(df)).select(\"prediction-label\", \"category\").take(20).foreach(println)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.ml.feature.IndexToString"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "name": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}