spark-1m.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 769.423095703125,
      "end_time": 1574546897818.095
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>17</td><td>application_1574540295268_0007</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://hn0-bdclus.fzt4pnr50moevdj1yjt0tv2y1b.bx.internal.cloudapp.net:8088/proxy/application_1574540295268_0007/\">Link</a></td><td><a target=\"_blank\" href=\"http://wn4-bdclus.fzt4pnr50moevdj1yjt0tv2y1b.bx.internal.cloudapp.net:30060/node/containerlogs/container_e05_1574540295268_0007_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n",
      "import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics"
     ]
    }
   ],
   "source": [
    "// Imports and starting the spark session.\n",
    "import org.apache.spark.mllib.regression.LabeledPoint\n",
    "import org.apache.spark.mllib.linalg.{Vector, Vectors}\n",
    "import org.apache.spark.mllib.tree.RandomForest\n",
    "import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 2289.953125,
      "end_time": 1574546905193.326
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "d_test: org.apache.spark.rdd.RDD[org.apache.spark.mllib.regression.LabeledPoint] = MapPartitionsRDD[5] at map at <console>:30"
     ]
    }
   ],
   "source": [
    "//Importing the train and test datasets from the csv files stored on 'azure storage' in the created cluster.(1m dataset)\n",
    "\n",
    "def csv_to_sparse_labpoint(fname:String) : org.apache.spark.rdd.RDD[LabeledPoint] = {\n",
    "  val rdd = sc.textFile(fname).map({ line =>\n",
    "    val vv = line.split(',').map(_.toDouble)\n",
    "    val label = vv(0) \n",
    "    val X = vv.slice(1,vv.size)\n",
    "    val n = X.filter(_!=0).length\n",
    "    var X_ids = Array.fill(n){0}\n",
    "    var X_vals = Array.fill(n){0.0}\n",
    "    var kk = 0\n",
    "    for( k <- 0 to X.length-1) {\n",
    "      if (X(k)!=0) {\n",
    "        X_ids(kk) = k\n",
    "        X_vals(kk) = X(k)\n",
    "        kk = kk + 1\n",
    "      }\n",
    "    }\n",
    "    val features = Vectors.sparse(X.length, X_ids, X_vals) \n",
    "    LabeledPoint(label, features)\n",
    "  })\n",
    "  return rdd\n",
    "}\n",
    "\n",
    "val d_train_0 = csv_to_sparse_labpoint(\"wasb://bdclusterstore@bdclusterhdistorage1.blob.core.windows.net/csvFiles/spark-train-1m.csv\")\n",
    "val d_test = csv_to_sparse_labpoint(\"wasb://bdclusterstore@bdclusterhdistorage1.blob.core.windows.net/csvFiles/spark-test-1m.csv\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 47509.580078125,
      "end_time": 1574546954057.054
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res9: Long = 1000000"
     ]
    }
   ],
   "source": [
    "// partition and caching the test and train dataset.\n",
    "d_train_0.partitions.size\n",
    "val d_train = d_train_0.repartition(32)\n",
    "d_train.partitions.size\n",
    "\n",
    "d_train.cache()\n",
    "d_test.cache()\n",
    "\n",
    "d_test.count()\n",
    "d_train.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 299259.291015625,
      "end_time": 1574547256396.096
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken is: 296.173986105"
     ]
    }
   ],
   "source": [
    "/* Model Training and Calculating the time taken for model training. \n",
    "# Parameters defined are:\n",
    "##  number_of_trees       = 100\n",
    "##  max_bins              = 20\n",
    "##  max_depth             = 50\n",
    "##  impurity              = gini \n",
    "##  featureSubsetStrategy = \"sqrt\" */\n",
    "\n",
    "val numClasses = 2\n",
    "val categoricalFeaturesInfo = Map[Int, Int]()\n",
    "val numTrees = 100\n",
    "val featureSubsetStrategy = \"sqrt\" \n",
    "val impurity = \"gini\"\n",
    "val maxDepth = 20     \n",
    "val maxBins = 50\n",
    "\n",
    "val now = System.nanoTime\n",
    "val model = RandomForest.trainClassifier(d_train, numClasses, categoricalFeaturesInfo,\n",
    "  numTrees, featureSubsetStrategy, impurity, maxDepth, maxBins)\n",
    "val total_time = ( System.nanoTime - now )/1e9\n",
    "println(\"Time taken is: \"+total_time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 19495.681884765625,
      "end_time": 1574547291966.922
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "res14: Double = 0.7072340383530936"
     ]
    }
   ],
   "source": [
    "// Prediction and AUC calculation for the model.\n",
    "\n",
    "import org.apache.spark.mllib.tree.configuration.FeatureType.Continuous\n",
    "import org.apache.spark.mllib.tree.model.{DecisionTreeModel, Node}\n",
    "def softPredict(node: Node, features: Vector): Double = {\n",
    "  if (node.isLeaf) {\n",
    "    if (node.predict.predict == 1.0) node.predict.prob else 1.0 - node.predict.prob\n",
    "  } else {\n",
    "    if (node.split.get.featureType == Continuous) {\n",
    "      if (features(node.split.get.feature) <= node.split.get.threshold) {\n",
    "        softPredict(node.leftNode.get, features)\n",
    "      } else {\n",
    "        softPredict(node.rightNode.get, features)\n",
    "      }\n",
    "    } else {\n",
    "      if (node.split.get.categories.contains(features(node.split.get.feature))) {\n",
    "        softPredict(node.leftNode.get, features)\n",
    "      } else {\n",
    "        softPredict(node.rightNode.get, features)\n",
    "      }\n",
    "    }\n",
    "  }\n",
    "}\n",
    "def softPredict2(dt: DecisionTreeModel, features: Vector): Double = {\n",
    "  softPredict(dt.topNode, features)\n",
    "}\n",
    "\n",
    "\n",
    "val scoreAndLabels = d_test.map { point =>\n",
    "  val score = model.trees.map(tree => softPredict2(tree, point.features)).sum / model.numTrees\n",
    "  (score, point.label)\n",
    "}\n",
    "val metrics = new BinaryClassificationMetrics(scoreAndLabels)\n",
    "metrics.areaUnderROC()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 3270.685791015625,
      "end_time": 1574547300310.767
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "predictionAndLabels: org.apache.spark.rdd.RDD[(Double, Double)] = MapPartitionsRDD[94] at map at <console>:33"
     ]
    }
   ],
   "source": [
    "// mapping the predict for other metric calculations.\n",
    "val predictionAndLabels = d_test.map { case LabeledPoint(label, features) =>\n",
    "     val prediction = model.predict(features)\n",
    "    (prediction, label)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 33607.965087890625,
      "end_time": 1574547333932.365
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Accuracy = 0.78515"
     ]
    }
   ],
   "source": [
    "val accuracy = predictionAndLabels.filter(x => x._1 == x._2).count().toDouble / predictionAndLabels.count().toDouble\n",
    "println(\"Accuracy = \" + accuracy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 17501.410888671875,
      "end_time": 1574547351444.701
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Error = 0.21485"
     ]
    }
   ],
   "source": [
    "val testErr = predictionAndLabels.filter(r => r._1 != r._2).count.toDouble / d_test.count()\n",
    "println(\"Test Error = \" + testErr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "cell_status": {
     "execute_time": {
      "duration": 17352.489013671875,
      "end_time": 1574547368808.924
     }
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Mean Squared Error = 0.21484999999999888"
     ]
    }
   ],
   "source": [
    "val testMSE = predictionAndLabels.map{ case(v, p) => math.pow((v - p), 2)}.mean()\n",
    "println(\"Test Mean Squared Error = \" + testMSE)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark",
   "language": "",
   "name": "sparkkernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}