als.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# In this notebook, we'll take some first steps with MLlib, Spark's machine learning library\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "http://spark.apache.org/docs/latest/mllib-guide.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark has two ML libraries to enable machine learning at scale:\n",
    "### It divides into two packages:\n",
    "\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."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We will try to predict user ratings based on other user ratings (collaborative filtering!). More specifically, we will:\n",
    "* Load user ratings data,\n",
    "* Separate it into training and test sets,\n",
    "* Train an Alternating Lease Squares model,\n",
    "* Make predictions on the test set\n",
    "* Compare the prediction to truth,\n",
    "* Produce a quantitative goodness-of-fit "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.recommendation import ALS\n",
    "from pyspark.mllib.recommendation import Rating"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "CORES_PER_NODE = 2\n",
    "NUM_WORKERS = 4\n",
    "REP_FACTOR = 4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create a DataFrame from the csv while repartitioning and persisting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Read in the ratings file (fromUserId, toUserId, rating).  These ratings are 0-9.\n",
    "ratings_raw_DF = sqlContext.read.format(\"com.databricks.spark.csv\") \\\n",
    "                           .options(header=\"false\") \\\n",
    "                           .load(\"s3n://insight-spark-after-dark/ratings.csv.gz\") \\\n",
    "                           .repartition(CORES_PER_NODE*NUM_WORKERS*REP_FACTOR)\\\n",
    "                           .persist(StorageLevel.MEMORY_AND_DISK_SER)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To write SparkSQL, we need to create a table object from our dataframe which we can use to run SparkSQL commands. The transformation registerTempTable() does this and we call our table 'ratings_raw_tbl'."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Register the ratings_raw DataFrame as a temp table - this allows us to run the SparkSQL queries\n",
    "ratings_raw_DF.registerTempTable(\"ratings_raw_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#let's trigger the job and load the data\n",
    "#just take a sample for now to save time\n",
    "#ratings_raw_DF = ratings_raw_DF.sample(False, .05, 20)\n",
    "ratings_raw_DF.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#let's look at it! (fromUserId, toUserId, rating)\n",
    "ratings_raw_DF.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Cast the DataFrame to enforce a schema with (from_user_id, to_user_id, rating)\n",
    "ratings_DF = sqlContext.sql(\"\"\"\n",
    "SELECT\n",
    "    CAST(C0 as int) AS from_user_id,\n",
    "    CAST(C1 as int) AS to_user_id,\n",
    "    CAST(C2 as int) AS rating\n",
    "FROM \n",
    "    ratings_raw_tbl\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create mllib.recommendation.Rating RDD from ratings DataFrame\n",
    "ratings_RDD = ratings_DF.rdd.map(lambda r: Rating(r.from_user_id, r.to_user_id, r.rating))\n",
    "\n",
    "ratings_RDD.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 1: Separate ratings data into training data (80%) and test data (20%)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Task 1: Separate ratings data into training data (80%) and test data (20%)\n",
    "split_ratings_RDD = ratings_RDD.randomSplit([0.8, 0.2])\n",
    "train_ratings_RDD = split_ratings_RDD[0]\n",
    "test_ratings_RDD = split_ratings_RDD[1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## One algorithm to learn the features in a collaborative filtering model is Alternating Least Squares. We learn factor vectors for each from_user and each to_user. Reach more here:\n",
    "#### http://bugra.github.io/work/notes/2014-04-19/alternating-least-squares-method-for-collaborative-filtering/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Train the ALS model using the training data and various model hyperparameters\n",
    "# hyperparameters are: \n",
    "# rank       number of features to use\n",
    "# iterations number of iterations of ALS (recommended: 10-20\n",
    "# lambda_     regularization factor (recommended: 0.01)\n",
    "# blocks     level of parallelism to split computation into\n",
    "# seed       random seed\n",
    "\n",
    "model = ALS.train(train_ratings_RDD, 1, iterations=5, lambda_=0.01, blocks=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Convert known test data to have only (from_user_id, to_user_id), \n",
    "# we can then feed this into our model to predict new ratings for \n",
    "# pairs of from_user_ids and to_user_ids\n",
    "test_from_to_RDD = test_ratings_RDD.map(lambda r: (r[0], r[1]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 2: Test the model by predicting the ratings for the test dataset\n",
    "#### Hint: the predictAll() method can be called with the RDD as input to return a new RDD\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 3: Prepare the known test predictions and actual predictions for comparison keyed by (from, to)\n",
    "#### Hint: We'd like to have an RDD with the form of ((from, to), rating), think map().\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 4: Join the known test predictions with the actual predictions\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 5: Evaluate the model using Mean Absolute Error (MAE) between the known test ratings and the actual predictions \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Next Steps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Task 6: Instead of a training and test set only, construct a cross validation set and use it to optimize over a vector of possible regularization parameters\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Task 7: If you're considering applying machine learning at scale for tomorrow's project, explore MLlib and consider your options!\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}