Spark demo #1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.environ[\"PYSPARK_SUBMIT_ARGS\"]='--packages com.databricks:spark-csv_2.10:1.1.0 pyspark-shell'\n",
    "os.environ[\"SPARK_HOME\"]='/home/cheshire/spark/'\n",
    "\n",
    "import sys\n",
    "\n",
    "sys.path.append(os.environ.get('SPARK_HOME', None)+\"/python\")\n",
    "sys.path.append(os.environ.get('SPARK_HOME', None)+\"/python/lib/py4j-0.8.2.1-src.zip\")\n",
    "\n",
    "import py4j\n",
    "from pyspark import SparkContext,SparkConf,SQLContext\n",
    "\n",
    "conf = (SparkConf().setMaster(\"local[2]\")\n",
    "        .setAppName(\"ML demo\")\n",
    "        .set(\"spark.executor.memory\", \"2g\")\n",
    "        .set(\"spark.cores.max\", \"2\"))\n",
    "\n",
    "sc = SparkContext(conf=conf)\n",
    "\n",
    "sqlCtx = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "u'1.5.0'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.regression import LabeledPoint, LinearRegressionWithSGD\n",
    "from numpy import array\n",
    "import random\n",
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "points = (sc.parallelize(range(100000))\n",
    "   .map(lambda x: LabeledPoint(x*25+17+random.uniform(-10,10), [float(x)])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[LabeledPoint(2034737.08514, [81389.0]),\n",
       " LabeledPoint(406994.669373, [16279.0]),\n",
       " LabeledPoint(293826.072539, [11752.0]),\n",
       " LabeledPoint(1217683.60407, [48707.0]),\n",
       " LabeledPoint(1001757.769, [40070.0]),\n",
       " LabeledPoint(1905722.73116, [76228.0]),\n",
       " LabeledPoint(1654993.16047, [66199.0]),\n",
       " LabeledPoint(1710859.82861, [68434.0]),\n",
       " LabeledPoint(1604484.89625, [64179.0]),\n",
       " LabeledPoint(414690.048177, [16587.0])]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "points.takeSample(False, 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "x = points.map(lambda p: p.features)\n",
    "y = points.map(lambda p: p.label)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DenseVector([0.0]),\n",
       " DenseVector([1.0]),\n",
       " DenseVector([2.0]),\n",
       " DenseVector([3.0]),\n",
       " DenseVector([4.0])]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[21.283261109702135,\n",
       " 50.75498884962461,\n",
       " 59.11946298258715,\n",
       " 101.15115966717568,\n",
       " 115.14982690022691]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "y.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "numIters = 20\n",
    "w = array([0])\n",
    "gamma = 1e-15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 [ 8.33329329] [-8.33329328683e+15]\n",
      "1 [ 13.88886381] [-5.55557052421e+15]\n",
      "2 [ 17.59260546] [-3.70374164507e+15]\n",
      "3 [ 20.06178496] [-2.46917950421e+15]\n",
      "4 [ 21.70791697] [-1.64613201214e+15]\n",
      "5 [ 22.80534652] [-1.09742954672e+15]\n",
      "6 [ 23.53697204] [-7.31625517382e+14]\n",
      "7 [ 24.02472603] [-4.87753994351e+14]\n",
      "8 [ 24.34989757] [-3.25171540389e+14]\n",
      "9 [ 24.56668047] [-2.16782897158e+14]\n",
      "10 [ 24.71120361] [-1.44523143566e+14]\n",
      "11 [ 24.80755274] [-9.63491262522e+13]\n",
      "12 [ 24.87178597] [-6.42332280374e+13]\n",
      "13 [ 24.91460851] [-4.28225398808e+13]\n",
      "14 [ 24.94315674] [-2.85482379429e+13]\n",
      "15 [ 24.96218905] [-1.9032310625e+13]\n",
      "16 [ 24.97487725] [-1.26881971057e+13]\n",
      "17 [ 24.98333621] [-8.45896040766e+12]\n",
      "18 [ 24.98897585] [-5.63963610556e+12]\n",
      "19 [ 24.99273538] [-3.75952726609e+12]\n"
     ]
    }
   ],
   "source": [
    "for i in range(numIters):\n",
    "    grad = x.zip(y).map(lambda (x,y): (w.dot(x)-y)*x).sum()\n",
    "    w = w - gamma*grad\n",
    "    print i, w, grad"
   ]
  },
  {
   "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}