PySpark_MLlib.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import matplotlib\n",
    "import seaborn as sns\n",
    "matplotlib.rcParams['savefig.dpi'] = 144"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark MLlib\n",
    "<!-- requirement: small_data/gutenberg -->\n",
    "*Official documentation [here](https://spark.apache.org/docs/latest/mllib-guide.html).*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext(\"local[*]\", \"temp\")\n",
    "print sc.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# needed to convert RDDs into DataFrames\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Tells Spark to look on the local filesystem\n",
    "import os\n",
    "def localpath(path):\n",
    "    return 'file://' + str(os.path.abspath(os.path.curdir)) + '/' + path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Algorithms\n",
    "\n",
    "Spark supports a number of machine-learning algorithms.\n",
    "\n",
    "- Classification and Regression\n",
    "    - SVM, linear regression\n",
    "    - SVR, logistic regression\n",
    "    - Naive Bayes\n",
    "    - Decision Trees\n",
    "    - Random Forests and Gradient-Boosted Trees\n",
    "- Clustering\n",
    "    - K-means (and streaming K-means)\n",
    "    - Gaussian Mixture Models\n",
    "    - Latent Dirichlet Allocation\n",
    "- Dimensionality Reduction\n",
    "    - SVD and PCA\n",
    "- It also has support for lower-level optimization primitives:\n",
    "    - Stochastic Gradient Descent\n",
    "    - Low-memory BFGS and L-BFGS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parallelized SGD\n",
    "\n",
    "For linear models like SVM, Linear Regression, and Logistic Regression, the cost function we're trying to optimize is essentially an average over the individual error term from each data point. This is particularly great for parallelization.  For example, in linear regression, recall that the gradient is\n",
    "\n",
    "$$\\begin{align}\n",
    "\\frac{\\partial \\log(L(\\beta))}{\\partial \\beta} &= \\frac{\\partial}{\\partial \\beta} \\frac{1}{2}\\sum_j \\|y_j - X_{j \\cdot} \\cdot \\beta\\| \\\\\n",
    "&= \\frac{1}{2}\\sum_j \\frac{\\partial}{\\partial \\beta} \\|y_j - X_{j \\cdot} \\cdot \\beta\\| \\\\\n",
    "& = \\sum_j y_j - X_{j \\cdot} \\cdot \\beta \\\\\n",
    "& \\approx \\sum_{sample \\mbox{ } j} y_j - X_{j \\cdot} \\cdot \\beta\n",
    "\\end{align}$$\n",
    "\n",
    "The key *mathematical properties* we have used are:\n",
    "\n",
    "1. the error functions are the sum of error contributions of different training instances\n",
    "1. linearity of the derivative\n",
    "1. associativity of addition\n",
    "1. downsampling giving an unbiased estimator\n",
    "\n",
    "Since the last sum is over the different training instances and these are stored on different nodes, we can parallelize the computation of the gradient in SGD across multiple nodes.  Of course, we still need to maintain the running weight $\\beta$ that has to be present on every node (through a broadcast variable that is updated).  Notice that SVM, Linear Regression, and Logistic Regression all have error functions that are just sums over training instances so SGD can be used for all these algorithms.\n",
    "\n",
    "Spark's [implementation](http://spark.apache.org/docs/latest/mllib-optimization.html#stochastic-gradient-descent-sgd) uses a tunable minibatch size parameter to sample a percentage of the features RDD. For each iteration, the updated weights are broadcast to the executors, and the update is calculated for each data point and sent back to be aggregated.\n",
    "\n",
    "Parallelization handles increasing number of sampled data points m quite well since there are no interaction terms and each calculation is independent. Controlling how the algorithm iterates to convergence is also important, and can be done with parameters for the total iterations and step size."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ML vs. MLlib packages\n",
    "\n",
    "Confusingly, there are two machine learning APIs in Spark, the `mllib` package based on RDDs and the `ml` package based on DataFrames. For years these have been developed somewhat in parallel, resulting in duplication and asymmetry in functionality.\n",
    "\n",
    "With Spark 2.0+, `mllib` is in maintenance mode and will likely be deprecated in future in favor of the DataFrame-based API which more closely resembles libraries like Scikit-learn. Below is one example of the RDD-based API; the rest of the notebook will focus on DataFrames."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.regression import LinearRegressionWithSGD, LinearRegressionModel, LabeledPoint\n",
    "from pyspark.mllib.evaluation import RegressionMetrics\n",
    "from pyspark.mllib.linalg import Vector, Vectors\n",
    "import random\n",
    "\n",
    "# parameters\n",
    "TRAINING_ITERATIONS = 10\n",
    "TRAINING_FRACTION = 0.6\n",
    "\n",
    "# generate the data\n",
    "data = sc.parallelize(xrange(1,10001)) \\\n",
    "    .map(lambda x: LabeledPoint(random.random(), [random.random(), random.random(), random.random()]))\n",
    "\n",
    "# split the training and test sets\n",
    "splits = data.randomSplit([TRAINING_FRACTION, 1 - TRAINING_FRACTION], seed=42)\n",
    "training, test = (splits[0].cache(), splits[1])\n",
    "\n",
    "# train the model\n",
    "model = LinearRegressionWithSGD.train(training, iterations=TRAINING_ITERATIONS)\n",
    "\n",
    "# get r2 score\n",
    "predictions = test.map(lambda x: (float(model.predict(x.features)), x.label))\n",
    "print RegressionMetrics(predictions).r2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Maybe we can improve this by modeling the intercept. Use `<shift-tab>` inside the arguments to bring up the docstring for LinearRegressionWithSGD, and rerun the training with an intercept term.\n",
    "\n",
    "If you're interested in methods for introspecting some of these objects, the `<tab>` and `<shift-tab>` documentation is good. You can also use `dir` in Python to list all the components of something."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dir(test.take(2)[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark ML\n",
    "Spark ML implements the ideas of transformers, estimators, and pipelines by standardizing APIS across machine learning algorithms. This can streamline more complex workflows.\n",
    "\n",
    "The core functionality includes:\n",
    "* DataFrames - built off Spark SQL, can be created either directly or from RDDs as seen above\n",
    "* Transformers - algorithms that accept a DataFrame as input and return a DataFrame as output\n",
    "* Estimators - algorithms that accept a DataFrame as input and return a Transformer as output\n",
    "* Pipelines - chaining together Transformers and Estimators\n",
    "* Parameters - common API for specifying hyperparameters\n",
    "\n",
    "For example, a learning algorithm can be implemented as an Estimator which trains on a DataFrame of features and returns a Transformer which can output predictions based on a test DataFrame.\n",
    "\n",
    "Full documentation can be found [here](http://spark.apache.org/docs/latest/ml-guide.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.feature import HashingTF, Tokenizer\n",
    "\n",
    "reviews = [(\"Prose is well-written, but style is an impediment to learning. Should be called 'Reviewing Spark,' not 'Learning Spark'\", 0.0),\n",
    "            (\"Nice Headstart to Spark\", 1.0),\n",
    "            (\"Start here: Excellent reference for Spark\", 1.0),\n",
    "            (\"Insightful and so Spark-tastic!\", 1.0),\n",
    "            (\"Good intro but wordy and lacking details in areas\", 0.0),\n",
    "            (\"Best of the Books Currently Available\", 1.0),\n",
    "            (\"A good resource for people interested in learning Spark\", 1.0),\n",
    "            (\"Great Overview\", 1.0)]\n",
    "\n",
    "test_reviews = [(\"A decent guided tour of Spark and its major components.\", 0.0),\n",
    "                (\"10/10 would buy again\", 1.0),\n",
    "                (\"it is simple to follow. well organized. straight ...\", 1.0),\n",
    "                (\"Just what you need to get started in Apache Spark.\", 1.0),\n",
    "                (\"Very good book for learning Spark\", 1.0)]\n",
    "\n",
    "training = sqlContext.createDataFrame(reviews, [\"title\", \"label\"]).cache()\n",
    "test = sqlContext.createDataFrame(test_reviews, [\"title\", \"label\"])\n",
    "\n",
    "tokenizer = Tokenizer(inputCol=\"title\", outputCol=\"words\")\n",
    "hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol=\"features\")\n",
    "logreg = LogisticRegression(maxIter=10, regParam=0.01)\n",
    "\n",
    "tokens = tokenizer.transform(training)\n",
    "hashes = hashingTF.transform(tokens)\n",
    "model = logreg.fit(hashes)\n",
    "\n",
    "# Make predictions on test documents\n",
    "test_tokens = tokenizer.transform(test)\n",
    "test_hashes = hashingTF.transform(test_tokens)\n",
    "\n",
    "prediction = model.transform(test_hashes)\n",
    "selected = prediction.select(\"title\", \"label\", \"prediction\")\n",
    "for row in selected.collect():\n",
    "    print(row)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Note that if you use a PipelineModel it won't have a coefficients attribute.\n",
    "model.coefficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "prediction.select([\"features\", \"probability\", \"prediction\"]).show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Exercise**: Rewrite the above using a Pipeline."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cross-validation and grid search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pipeline = Pipeline(stages=[tokenizer, hashingTF, logreg])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "paramGrid = ParamGridBuilder() \\\n",
    "    .addGrid(hashingTF.numFeatures, [10, 100, 1000]) \\\n",
    "    .addGrid(logreg.regParam, [0.1, 0.01]) \\\n",
    "    .build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "crossval = CrossValidator(estimator=pipeline,\n",
    "                          estimatorParamMaps=paramGrid,\n",
    "                          evaluator=BinaryClassificationEvaluator(),\n",
    "                          numFolds=3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "*Note*: A more traditional validation set without folding is available in `TrainValidationSplit`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "cvModel = crossval.fit(training)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "better_prediction = cvModel.transform(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "selected = better_prediction.select(\"title\", \"label\", \"prediction\")\n",
    "for row in selected.collect():\n",
    "    print(row)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "cvModel.bestModel.stages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "cvModel.bestModel.stages[2].coefficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "training.unpersist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example algorithm: Word2Vec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Word2Vec\n",
    "\n",
    "# text = sc.parallelize(reviews + test_reviews).map(lambda (line, score): (line.split(\" \"), score)).toDF(['text', 'score'])\n",
    "gutenberg = sc.textFile(localpath('small_data/gutenberg/')).map(lambda line: (line.split(\" \"), 1)).toDF(['text', 'score'])\n",
    "w2v = Word2Vec(inputCol=\"text\", outputCol=\"vectors\")\n",
    "model = w2v.fit(gutenberg)\n",
    "result = model.transform(gutenberg)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vectors = model.getVectors().rdd.map(lambda x: (x.word, x.vector))\n",
    "print model.findSynonyms('woman', 10).rdd.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "king_vec = vectors.lookup('king')[0]\n",
    "queen_vec = vectors.lookup('queen')[0]\n",
    "man_vec = vectors.lookup('man')[0]\n",
    "woman_vec = vectors.lookup('woman')[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print king_vec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print queen_vec.squared_distance(king_vec)\n",
    "print queen_vec.squared_distance(woman_vec)\n",
    "print queen_vec.squared_distance(man_vec)\n",
    "print queen_vec.squared_distance(king_vec + man_vec - woman_vec)\n",
    "print queen_vec.squared_distance(king_vec - man_vec + woman_vec)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature processing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vdf = vectors.toDF([\"word\", \"vector\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vdf.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sample_vector = vdf.select(\"vector\").take(1)[0][0]\n",
    "print len(sample_vector)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorSlicer\n",
    "\n",
    "first_slicer = VectorSlicer(inputCol=\"vector\", outputCol=\"first\", indices=[0])\n",
    "last_slicer = VectorSlicer(inputCol=\"vector\", outputCol=\"last\", indices=[len(sample_vector) - 1])\n",
    "med_slicer = VectorSlicer(inputCol=\"vector\", outputCol=\"med\", indices=range(45, 55))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "output = med_slicer.transform(last_slicer.transform(first_slicer.transform(vdf)))\n",
    "output.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "assembler = VectorAssembler(\n",
    "    inputCols=[\"first\", \"last\", \"med\"],\n",
    "    outputCol=\"features\")\n",
    "\n",
    "new_output = assembler.transform(output)\n",
    "new_output.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Binarizer\n",
    "\n",
    "binarizer = Binarizer(threshold=0.05, inputCol=\"features\", outputCol=\"bin_features\")\n",
    "new_output_b = binarizer.transform(new_output)\n",
    "print new_output_b.select([\"features\", \"bin_features\"]).take(1)[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: Use SVM to predict colon cancer from gene expressions\n",
    "You can start getting a feel for the MLlib operations by following the [Spark docs example](https://spark.apache.org/docs/1.3.0/mllib-linear-methods.html#linear-support-vector-machines-svms) on this dataset.\n",
    "\n",
    "#### About the data format: LibSVM\n",
    "MLlib conveniently provides a data loading method, `MLUtils.loadLibSVMFile()`, for the LibSVM format for which many other languages (R, Matlab, etc.) also have loading methods.  \n",
    "A dataset of *n* features will have one row per datum, with the label and values of each feature organized as follows:\n",
    ">{label} 1:{value} 2:{value} ... n:{value}\n",
    "\n",
    "Take these two datapoints with six features and labels of -1 and 1 respectively as an example:\n",
    ">-1.000000  1:2.080750 2:1.099070 3:0.927763 4:1.029080 5:-0.130763 6:1.265460  \n",
    "1.000000  1:1.109460 2:0.786453 3:0.445560 4:-0.146323 5:-0.996316 6:0.555759 \n",
    "\n",
    "#### About the colon-cancer dataset\n",
    "This dataset was introduced in the 1999 paper \"Broad patterns of gene expression revealed by clustering analysis of tumor and normal colon tissues probed by oligonucleotide arrays.\" (Available on PNAS)\n",
    "\n",
    "Here's the abstract of the paper:  \n",
    "> *Oligonucleotide arrays can provide a broad picture of the state of the cell, by monitoring the expression level of thousands of genes at the same time. It is of interest to develop techniques for extracting useful information from the resulting data sets. Here we report the application of a two-way clustering method for analyzing a data set consisting of the expression patterns of different cell types. Gene expression in 40 tumor and 22 normal colon tissue samples was analyzed with an Affymetrix oligonucleotide array complementary to more than 6,500 human genes. An efficient two-way clustering algorithm was applied to both the genes and the tissues, revealing broad coherent patterns that suggest a high degree of organization underlying gene expression in these tissues. Coregulated families of genes clustered together, as demonstrated for the ribosomal proteins. Clustering also separated cancerous from noncancerous tissue and cell lines from in vivo tissues on the basis of subtle distributed patterns of genes even when expression of individual genes varied only slightly between the tissues. Two-way clustering thus may be of use both in classifying genes into functional groups and in classifying tissues based on gene expression.*\n",
    "\n",
    "There are 2000 features, 62 data points (40 tumor (label=0), 22 normal (label=1)), and 2 classes (labels) for the colon cancer dataset. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exit Tickets\n",
    "1. When would you use `org.apache.spark.mllib.linalg.Vector` versus `breeze.linalg.DenseVector`?\n",
    "1. Why can SVM, Linear Regression, and Logistic Regression be parallelized?  How would you parallelize KMeans?\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Copyright &copy; 2015 The Data Incubator.  All rights reserved.*"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}