Assignment 2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/bin/bash\n",
    "#$ -l h_rt=4:00:00  #time needed\n",
    "#$ -pe smp 20 #number of cores\n",
    "#$ -l rmem=15G #number of memery\n",
    "#$ -o COM6012_Assignment2.output #This is where your output and errors are logged.\n",
    "#$ -j y # normal and error outputs into a single file (the file above)\n",
    "#$ -m ea #Email you when it finished or aborted\n",
    "#$ -cwd # Run job from current directory\n",
    "#$ -P rse-com6012\n",
    "#$ -q rse-com6012.q\n",
    "\n",
    "import os\n",
    "import subprocess\n",
    "def module(*args):        \n",
    "    if isinstance(args[0], list):        \n",
    "        args = args[0]        \n",
    "    else:        \n",
    "        args = list(args)        \n",
    "    (output, error) = subprocess.Popen(['/usr/bin/modulecmd', 'python'] + args, stdout=subprocess.PIPE).communicate()\n",
    "    exec(output)    \n",
    "module('load', 'apps/java/jdk1.8.0_102/binary')    \n",
    "os.environ['PYSPARK_PYTHON'] = os.environ['HOME'] + '/.conda/envs/jupyter-spark/bin/python'\n",
    "\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[20]\") \\\n",
    "    .appName(\"Question 1\") \\\n",
    "    .config(\"spark.local.dir\",\"/fastdata/acp18dck\") \\\n",
    "    .config(\"spark.executor.memory\", \"15g\") \\\n",
    "    .config(\"spark.executor.cores\", \"20\") \\\n",
    "    .config(\"spark.driver.memory\", \"15g\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext\n",
    "os.environ['PATH'] = os.environ['HOME'] + '/.conda/envs/jupyter-spark/bin/'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 1.1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "schema = StructType([\n",
    "    StructField('label', DoubleType()), StructField('lepton_pT', DoubleType()), StructField('lepton_eta', DoubleType()),\n",
    "    StructField('lepton_phi', DoubleType()), StructField('missing_energy_mag', DoubleType()), StructField('missing_energy_phi', DoubleType()),\n",
    "    StructField('jet_1_pt', DoubleType()), StructField('jet_1_eta', DoubleType()), StructField('jet_1_phi', DoubleType()),\n",
    "    StructField('jet_1_b-tag', DoubleType()), StructField('jet_2_pt', DoubleType()), StructField('jet_2_eta', DoubleType()),\n",
    "    StructField('jet_2_phi', DoubleType()), StructField('jet_2_b-tag', DoubleType()), StructField('jet_3_pt', DoubleType()),\n",
    "    StructField('jet_3_eta', DoubleType()), StructField('jet_3_phi', DoubleType()), StructField('jet_3_b-tag', DoubleType()),\n",
    "    StructField('jet_4_pt', DoubleType()), StructField('jet_4_eta', DoubleType()), StructField('jet_4_phi', DoubleType()),\n",
    "    StructField('jet_4_b-tag', DoubleType()), StructField('m_jj', DoubleType()), StructField('m_jjj', DoubleType()),\n",
    "    StructField('m_lv', DoubleType()), StructField('m_jlv', DoubleType()), StructField('m_bb', DoubleType()),\n",
    "    StructField('m_wbb', DoubleType()), StructField('m_wwbb', DoubleType())\n",
    "])\n",
    "\n",
    "## load the higgs dataset with defines schema\n",
    "\n",
    "higgsData = spark.read.load('Data/HIGGS.csv.gz', format = 'csv', sep = ',', schema = schema, header = 'false').cache()\n",
    "higgsSubset, remainder = higgsData.randomSplit([0.25, 0.75], 9)\n",
    "\n",
    "import time\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import VectorAssembler, Binarizer, HashingTF\n",
    "from pyspark.ml.classification import LogisticRegression, DecisionTreeClassifier\n",
    "from pyspark.ml.regression import DecisionTreeRegressor\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator, BinaryClassificationEvaluator\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "\n",
    "schemaNames = higgsSubset.schema.names\n",
    "ncolumns = len(higgsSubset.columns)\n",
    "\n",
    "## Vectorise the features and full dataset into training and test set\n",
    "\n",
    "assembler = VectorAssembler(inputCols = schemaNames[1: ncolumns - 1], outputCol = 'features')\n",
    "subsetVector = assembler.transform(higgsSubset)\n",
    "higgsSubsetData = subsetVector.select('label', 'features')\n",
    "trainingData, testData = higgsSubsetData.randomSplit([0.7, 0.3], 11)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtcStart = time.time()\n",
    "\n",
    "##### Decision Tree Classifier ##### \n",
    "\n",
    "decisionTreeClassifier = DecisionTreeClassifier(labelCol = 'label', featuresCol = 'features', maxDepth = 10, impurity = 'entropy')\n",
    "dtcPipe = Pipeline(stages = [decisionTreeClassifier])\n",
    "dtcEvaluator = MulticlassClassificationEvaluator(labelCol = 'label', predictionCol = 'prediction', metricName = 'accuracy')\n",
    "dtcGrid = ParamGridBuilder().addGrid(32, [5, 10, 20]).build()\n",
    "dtcValidator = CrossValidator(estimator = dtcPipe, estimatorParamMaps = dtcGrid, evaluator = dtcEvaluator, numFolds = 5)\n",
    "\n",
    "dtcModel = dtcValidator.fit(trainingData)\n",
    "print('List of Features and their Importance for Decision Tree Classifier')\n",
    "print(dtcModel.bestModel.stages[-1].featureImportances)\n",
    "\n",
    "dtcEnd = time.time()\n",
    "print(dtcEnd - dtcStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtrStart = time.time()\n",
    "\n",
    "##### Decision Tree Regressor ##### \n",
    "\n",
    "decisionTreeRegressor = DecisionTreeRegressor(labelCol = 'label', featuresCol = 'features', predictionCol = 'predNonBinary', maxDepth = 10)\n",
    "binarizer = Binarizer(threshold = 0.5, inputCol = 'predNonBinary', outputCol = 'prediction')\n",
    "dtrPipe = Pipeline(stages = [decisionTreeRegressor, binarizer])\n",
    "dtrEvaluator = BinaryClassificationEvaluator(labelCol = 'label', rawPredictionCol = 'prediction', metricName = 'areaUnderROC')\n",
    "dtrGrid = ParamGridBuilder().addGrid(32, [5, 10, 20]).build()\n",
    "dtrValidator = CrossValidator(estimator = dtrPipe, estimatorParamMaps = dtrGrid, evaluator = dtrEvaluator, numFolds = 5)\n",
    "\n",
    "dtrModel = dtrValidator.fit(trainingData)\n",
    "print('List of Features abd their Importance for Decision Tree Regessor')\n",
    "print(dtrModel.bestModel.stages[0].featureImportances)\n",
    "\n",
    "dtrEnd = time.time()\n",
    "print(dtrEnd - dtrStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "lrStart = time.time()\n",
    "\n",
    "##### Logistic Regression Classifier #####\n",
    "\n",
    "lr = LogisticRegression(labelCol = 'label', featuresCol = 'features', maxIter = 10, regParam = 0.01, family = 'binomial')\n",
    "lrPipe = Pipeline(stages = [lr])\n",
    "lrEvaluator = MulticlassClassificationEvaluator(labelCol = 'label', predictionCol = 'prediction', metricName = 'accuracy')\n",
    "lrGrid = ParamGridBuilder().addGrid(lr.regParam, [0.1, 0.01, 0.001]).build()\n",
    "lrValidator = CrossValidator(estimator = lrPipe, estimatorParamMaps = lrGrid, evaluator = lrEvaluator, numFolds = 5)\n",
    "\n",
    "lrModel = lrValidator.fit(trainingData)\n",
    "lrBestModel = lrModel.bestModel.stages[-1].coefficients\n",
    "print('List of Features and their absolute Importance for Logistic Regression')\n",
    "print(np.abs(lrBestModel))\n",
    "\n",
    "lrEnd = time.time()\n",
    "print(lrEnd - lrStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtcPreds = dtcModel.transform(testData)\n",
    "dtcAccuracy = dtcEvaluator.evaluate(dtcPreds) \n",
    "print('Decision Tree Classifier')\n",
    "print(\"Accuracy = %g \" % dtcAccuracy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtrPreds = dtrModel.transform(testData)\n",
    "dtrAccuracy = dtrEvaluator.evaluate(dtrPreds)\n",
    "print('Decision Tree Regressor')\n",
    "print(\"Accuracy = %g \" % dtrAccuracy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lrPreds = lrModel.transform(testData)\n",
    "lrAccuracy = lrEvaluator.evaluate(lrPreds)\n",
    "print('Logistic Regression')\n",
    "print(\"Accuracy = %g\" % lrAccuracy)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 1.2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fullDatasetVector = assembler.transform(higgsData)\n",
    "higgsData = fullDatasetVector.select('label', 'features')\n",
    "fullTrainingData, fullTestData = higgsData.randomSplit([0.7, 0.3], 11)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "##### Decision Tree Classifier ##### \n",
    "dtcFullstart = time.time()\n",
    "dtcModelFull = dtcValidator.fit(fullTrainingData)\n",
    "print('List of Features and their Importance for Decision Tree Classifier for full Dataset')\n",
    "print(dtcModelFull.bestModel.stages[-1].featureImportances)\n",
    "dtcFullEnd = time.time()\n",
    "\n",
    "print(dtcFullEnd - dtcFullstart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "##### Decision Tree Regressor ##### \n",
    "dtrFullStart = time.time()\n",
    "dtrModelFull = dtrValidator.fit(fullTrainingData)\n",
    "print('List of Features and their Importance for Decision Tree Regessor for full Dataset')\n",
    "print(dtrModelFull.bestModel.stages[0].featureImportances)\n",
    "dtrFullEnd = time.time()\n",
    "\n",
    "print(dtrFullEnd - dtrFullStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "##### Logistic Regression Classifier #####\n",
    "lrFullStart = time.time()\n",
    "lrModelFull = lrValidator.fit(fullTrainingData)\n",
    "print('List of Features and their absolute Importance for Logistic Regression')\n",
    "lrBestModelFull = lrModelFull.bestModel.stages[-1].coefficients\n",
    "print(np.abs(lrBestModelFull))\n",
    "lrFullEnd = time.time()\n",
    "\n",
    "print(lrFullEnd - lrFullStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtcFullPreds = dtcModelFull.transform(fullTestData)\n",
    "dtcAccuracyFull = dtcEvaluator.evaluate(dtcFullPreds)\n",
    "print('Decision Tree Classifier')\n",
    "print(\"Accuracy = %g \" % dtcAccuracyFull)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dtrFullPreds = dtrModelFull.transform(fullTestData)\n",
    "dtrAccuracyFull = dtrEvaluator.evaluate(dtrFullPreds)\n",
    "print('Decision Tree Regressor')\n",
    "print(\"Accuracy = %g \" % dtrAccuracyFull)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lrFullPreds = lrModelFull.transform(fullTestData)\n",
    "lrAccuracyFull = lrEvaluator.evaluate(lrFullPreds)\n",
    "print('Logistic Regression')\n",
    "print(\"Accuracy = %g\" % lrAccuracyFull)\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 2.1a"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!/bin/bash\n",
    "#$ -l h_rt=4:00:00  #time needed\n",
    "#$ -pe smp 20 #number of cores\n",
    "#$ -l rmem=15G #number of memery\n",
    "#$ -o COM6012_Assignment2.output #This is where your output and errors are logged.\n",
    "#$ -j y # normal and error outputs into a single file (the file above)\n",
    "#$ -M  #Notify you by email, remove this line if you don't like\n",
    "#$ -m ea #Email you when it finished or aborted\n",
    "#$ -cwd # Run job from current directory\n",
    "#$ -P rse-com6012\n",
    "#$ -q rse-com6012.q\n",
    "\n",
    "import os\n",
    "import subprocess\n",
    "def module(*args):        \n",
    "    if isinstance(args[0], list):        \n",
    "        args = args[0]        \n",
    "    else:        \n",
    "        args = list(args)        \n",
    "    (output, error) = subprocess.Popen(['/usr/bin/modulecmd', 'python'] + args, stdout=subprocess.PIPE).communicate()\n",
    "    exec(output)    \n",
    "module('load', 'apps/java/jdk1.8.0_102/binary')    \n",
    "os.environ['PYSPARK_PYTHON'] = os.environ['HOME'] + '/.conda/envs/jupyter-spark/bin/python'\n",
    "\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[20]\") \\\n",
    "    .appName(\"Question 2\") \\\n",
    "    .config(\"spark.local.dir\",\"/fastdata/acp18dck\") \\\n",
    "    .config(\"spark.executor.memory\", \"15g\") \\\n",
    "    .config(\"spark.executor.cores\", \"20\") \\\n",
    "    .config(\"spark.driver.memory\", \"15g\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext\n",
    "os.environ['PATH'] = os.environ['HOME'] + '/.conda/envs/jupyter-spark/bin/'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = spark.read.load('Data/train_set.csv', format = 'csv', sep = ',', inferSchema = 'true', header = 'true').cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import time\n",
    "\n",
    "schemaNames = data.schema.names\n",
    "dataCount = data.count()\n",
    "\n",
    "matrix = [[None for r in range(len(schemaNames))] for r in range(dataCount)]\n",
    "# list of lists with length data count each list has schemaNames length\n",
    "\n",
    "## Method where if the data doesn't exist the piece of data from a previous row where data did exist is used\n",
    "for x, column in enumerate(schemaNames):\n",
    "    rowCounter = 0\n",
    "    lastValue = None\n",
    "    for row in data.select(column).collect():\n",
    "        if row[column] is '?':\n",
    "            matrix[rowCounter][x] = lastValue\n",
    "        else:\n",
    "            lastValue = row[column]\n",
    "            matrix[rowCounter][x] = row[column]\n",
    "        rowCounter += 1\n",
    "\n",
    "matrix = matrix[1:]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 2.1b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, OneHotEncoderEstimator, Binarizer\n",
    "\n",
    "## Build the dataframe from the matrix where missing data has been filled\n",
    "\n",
    "dataframe = spark.createDataFrame(pd.DataFrame(matrix, columns = schemaNames))\n",
    "catIndexes = [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 29]\n",
    "others = [0, 1, 2, 3, 4, 20, 21, 22, 23, 24, 25, 26, 27, 28, 30, 31, 32, 33, 34]\n",
    "categoricalIndexes = np.array(schemaNames)[catIndexes]\n",
    "otherIndexes = np.array(schemaNames)[others]\n",
    "outputsIndexer = [str(value + '_ind') for value in categoricalIndexes]\n",
    "outputsEncoder = [str(value + '_enc') for value in categoricalIndexes]\n",
    "\n",
    "## Endcode all catagorical variables with one hot encoder, first requires string indexer to turn strings into integers\n",
    "\n",
    "indexers = [StringIndexer(inputCol = column, outputCol = column + '_ind', handleInvalid =  'skip').fit(dataframe) for column in categoricalIndexes]\n",
    "encoder = OneHotEncoderEstimator(inputCols = outputsIndexer, outputCols = outputsEncoder)\n",
    "indexers.append(encoder)\n",
    "encPipe = Pipeline(stages = indexers)\n",
    "encodedData = encPipe.fit(dataframe).transform(dataframe)\n",
    "\n",
    "requiredColumns = np.concatenate([otherIndexes, np.array(outputsEncoder)])\n",
    "requiredColumns = requiredColumns.tolist()\n",
    "\n",
    "encodedDataFrame = encodedData.select(requiredColumns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 2.1c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as func\n",
    "\n",
    "## Add a weights column to counter the highly unbalanced data, we want data that has claim information to be weighted highly\n",
    "## As there are few entries with information about claims\n",
    "\n",
    "hasClaimCount = encodedDataFrame.where(encodedDataFrame['Claim_Amount'] != 0.0).count()\n",
    "weightRatio = hasClaimCount / encodedDataFrame.count()\n",
    "\n",
    "encodedDataFrame = encodedDataFrame.withColumn('weights', func.when(encodedDataFrame['Claim_Amount'] == 0.0, weightRatio).otherwise(1.0 - weightRatio))    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 2.2a"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "typeChangeDF = encodedDataFrame.withColumn('OrdCat', encodedDataFrame['OrdCat'].cast(DoubleType()))\n",
    "## Need to re-run above two cells before this one\n",
    "inputColumns = requiredColumns + ['weights']\n",
    "inputColumns.remove('Claim_Amount')\n",
    "inputColumns.remove('Row_ID')\n",
    "inputColumns.remove('Household_ID')\n",
    "assembler = VectorAssembler(inputCols = inputColumns, outputCol = 'features')\n",
    "transformedDataFrame = assembler.transform(typeChangeDF)\n",
    "binariserP2 = Binarizer(threshold = weightRatio + 0.0001, inputCol = 'Claim_Amount', outputCol = 'label')\n",
    "binarizedDF = binariserP2.transform(transformedDataFrame)\n",
    "\n",
    "trainingData, testData = binarizedDF.randomSplit([0.75, 0.25], 9)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import GeneralizedLinearRegression\n",
    "lmFullStart = time.time()\n",
    "\n",
    "lm = GeneralizedLinearRegression(featuresCol = 'features', labelCol = 'Claim_Amount', weightCol = 'weights', maxIter = 10, regParam = 0.01, family = 'poisson', link = 'log')\n",
    "lmmodel = lm.fit(trainingData)\n",
    "lmPreds = lmmodel.transform(testData)\n",
    "\n",
    "lmFullStop = time.time()\n",
    "print(lmFullStop - lmFullStart)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "lmEvaluator = RegressionEvaluator(labelCol = 'Claim_Amount', predictionCol = 'prediction', metricName = 'rmse')\n",
    "lmRmse = lmEvaluator.evaluate(lmPreds)\n",
    "print(\"RMSE = %g \" % lmRmse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Assignment Question 2.2b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "lr = LogisticRegression(maxIter=100, regParam=0.9)\n",
    "pipelineLR = Pipeline(stages=[lr])\n",
    "modelLR = pipelinelR.fit(trainingData)\n",
    "print('Logistic those who claimed:')\n",
    "print('')\n",
    "modelLR.transform(testData).show(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import GeneralizedLinearRegression\n",
    "\n",
    "glr = GeneralizedLinearRegression(family=\"gamma\", link=\"identity\", maxIter=100, regParam=0.001)\n",
    "modelGLR = glr.fit(trainingData)"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python (36)",
   "language": "python",
   "name": "py36"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}