notebooks/picluster.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Notebook Walkthrough"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Before running be sure to have cluster and Spark history server started:\n",
    "#### start-dfs.sh && start-yarn.sh && $SPARK_HOME/sbin/start-history-server.sh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext, SQLContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create new SparkSession for this app and give it a name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('PySpark-100Node-ANN5').getOrCreate()\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create SQLContext within current SparkSession to read in and work with data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql_c = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read data in from the DFS, put the data in the HDFS using the below command.\n",
    "hdfs dfs -put data.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = 'data.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sql_c.read.csv(path, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+----------------+--------+--------+--------+--------+--------+-------------+-----+\n",
      "|               objid|              ra|             dec|       u|       g|       r|       i|       z|     redshift|class|\n",
      "+--------------------+----------------+----------------+--------+--------+--------+--------+--------+-------------+-----+\n",
      "|1.23766870461946E...|262.675447007966|7.03373899198224|19.19479|16.72763| 15.6987|15.32512|15.11311| 0.0001174448| STAR|\n",
      "|1.23766870515647E...|263.155622796506| 7.2609959506419|17.21147|16.16533|15.79078|15.62873|15.58098| 7.529317E-05| STAR|\n",
      "|1.23766870515627E...|262.749186164557|7.36335372355878|17.57057|16.54758|16.14557|15.96953|15.89596|-0.0008980818| STAR|\n",
      "|1.23766857147502E...|262.221092584887|6.94796600919322|22.68772|20.27666|19.59911|19.29291|19.32655|-0.0001748509| STAR|\n",
      "|1.2376687051566E+018|263.367549608197|7.02018938642791|19.29753|18.01646|17.43456|17.21292|17.05329|-0.0002282125| STAR|\n",
      "+--------------------+----------------+----------------+--------+--------+--------+--------+--------+-------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Display first 5 dataframe entries\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- objid: string (nullable = true)\n",
      " |-- ra: string (nullable = true)\n",
      " |-- dec: string (nullable = true)\n",
      " |-- u: string (nullable = true)\n",
      " |-- g: string (nullable = true)\n",
      " |-- r: string (nullable = true)\n",
      " |-- i: string (nullable = true)\n",
      " |-- z: string (nullable = true)\n",
      " |-- redshift: string (nullable = true)\n",
      " |-- class: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The above schema shows that the current columns stored in the dataframe are, by default, string data types.\n",
    "### We fix this issue here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to convert each dataframe column to the correct type\n",
    "def convertColumns(df, names, newTypes):\n",
    "    for name, newType in zip(names, newTypes): \n",
    "        df = df.withColumn(name, df[name].cast(newType))\n",
    "    return df "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "names = ['objid', 'ra', 'dec', 'u', 'g', 'r', 'i', 'z', 'redshift', 'class']\n",
    "types = [FloatType(), FloatType(), FloatType(), FloatType(), FloatType(), FloatType(), \n",
    "             FloatType(), FloatType(), FloatType(), StringType()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = convertColumns(df, names, types)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- objid: float (nullable = true)\n",
      " |-- ra: float (nullable = true)\n",
      " |-- dec: float (nullable = true)\n",
      " |-- u: float (nullable = true)\n",
      " |-- g: float (nullable = true)\n",
      " |-- r: float (nullable = true)\n",
      " |-- i: float (nullable = true)\n",
      " |-- z: float (nullable = true)\n",
      " |-- redshift: float (nullable = true)\n",
      " |-- class: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now each feature has its proper type. Next, we drop the logistical features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------+--------+--------+--------+------------+-----+\n",
      "|       u|       g|       r|       i|       z|    redshift|class|\n",
      "+--------+--------+--------+--------+--------+------------+-----+\n",
      "|19.19479|16.72763| 15.6987|15.32512|15.11311| 1.174448E-4| STAR|\n",
      "|17.21147|16.16533|15.79078|15.62873|15.58098| 7.529317E-5| STAR|\n",
      "|17.57057|16.54758|16.14557|15.96953|15.89596|-8.980818E-4| STAR|\n",
      "|22.68772|20.27666|19.59911|19.29291|19.32655|-1.748509E-4| STAR|\n",
      "|19.29753|18.01646|17.43456|17.21292|17.05329|-2.282125E-4| STAR|\n",
      "+--------+--------+--------+--------+--------+------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = df.drop(*['objid', 'ra', 'dec'])\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### All that remains now are the numerical features and the corresponding class.\n",
    "### The next step is to condense the input dataframe into a dataframe with two entries: A 'label' feature and a DenseVector 'features' containing the combined numerical data. PySpark uses dataframes with this format for the training of ML models."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.linalg import DenseVector"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_data = df.rdd.map(lambda x: (x[6], DenseVector(x[:6])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.createDataFrame(input_data, [\"label\", \"features\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+\n",
      "|label|            features|\n",
      "+-----+--------------------+\n",
      "| STAR|[19.1947898864746...|\n",
      "| STAR|[17.2114696502685...|\n",
      "| STAR|[17.5705699920654...|\n",
      "| STAR|[22.6877193450927...|\n",
      "| STAR|[19.2975292205810...|\n",
      "+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# New df has 2 columns\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(label='STAR', features=DenseVector([19.1948, 16.7276, 15.6987, 15.3251, 15.1131, 0.0001]))]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 'features' column contains all of the previously separate numeric data\n",
    "df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "| label| count|\n",
      "+------+------+\n",
      "|GALAXY|300045|\n",
      "|  STAR|124392|\n",
      "|   QSO| 75563|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Show label counts by class\n",
    "df.groupBy(\"label\") \\\n",
    "    .count() \\\n",
    "    .orderBy(col(\"count\").desc()) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The current data must now be transformed further to be used to train a ML model\n",
    "### First, we need to encode the label classes to be numeric. For this we use the StringIndexer function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StringIndexer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Using the label column, create a new\n",
    "stringIndexer = StringIndexer(inputCol=\"label\", outputCol=\"label_new\")\n",
    "indexer = stringIndexer.fit(df)\n",
    "scaled_df = indexer.transform(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Second, we standardize the numeric features by subtracting the mean and scale to unit variance each feature by using the StandardScaler function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StandardScaler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "standardScaler = StandardScaler(inputCol=\"features\", outputCol=\"features_scaled\")\n",
    "scaler = standardScaler.fit(scaled_df)\n",
    "scaled_df = scaler.transform(scaled_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(label='STAR', features=DenseVector([19.1948, 16.7276, 15.6987, 15.3251, 15.1131, 0.0001]), label_new=1.0, features_scaled=DenseVector([0.3105, 0.2781, 0.296, 0.2792, 0.2666, 0.0002]))]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Checkout new dataframe structure\n",
    "scaled_df.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Since we now have the newly processed features, we drop the original features and rename the new."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "scaled_df = scaled_df.drop('features')\n",
    "scaled_df = scaled_df.drop('label')\n",
    "scaled_df = scaled_df.withColumnRenamed(\"label_new\", \"label\")\n",
    "scaled_df = scaled_df.withColumnRenamed(\"features_scaled\", \"features\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+\n",
      "|label|            features|\n",
      "+-----+--------------------+\n",
      "|  1.0|[0.31051706979631...|\n",
      "|  1.0|[0.27843259313067...|\n",
      "|  1.0|[0.28424181462031...|\n",
      "|  1.0|[0.36702272716012...|\n",
      "|  1.0|[0.31217909981426...|\n",
      "+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# View the final processed dataframe\n",
    "scaled_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+\n",
      "|label| count|\n",
      "+-----+------+\n",
      "|  0.0|300045|\n",
      "|  1.0|124392|\n",
      "|  2.0| 75563|\n",
      "+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# The new numeric labels correspond to the same frequencies above, as expected\n",
    "scaled_df.groupBy(\"label\") \\\n",
    "    .count() \\\n",
    "    .orderBy(col(\"count\").desc()) \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model Building"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 100 Epoch Multilayer Perceptron"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import MultilayerPerceptronClassifier\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = scaled_df.randomSplit([.8,.2], seed=101)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training Dataset Count: 400117\n",
      "Test Dataset Count: 99883\n"
     ]
    }
   ],
   "source": [
    "print(\"Training Dataset Count: \" + str(train.count()))\n",
    "print(\"Test Dataset Count: \" + str(test.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "# specify layers for the neural network:\n",
    "# input layer of size 6 (features), three hidden layers of size 64, 64, 32\n",
    "# and output of size 3 (# classes)\n",
    "layers = [6, 64, 64, 32, 3]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "trainer = MultilayerPerceptronClassifier(maxIter=100, layers=layers, blockSize=128, seed=101)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Time model training time\n",
    "start = time.time()\n",
    "model = trainer.fit(train)\n",
    "duration = time.time() - start"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MLP training time: 1133.23s\n"
     ]
    }
   ],
   "source": [
    "print('MLP training time: {:.2f}s'.format(duration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "result = model.transform(test)\n",
    "predictionAndLabels = result.select(\"prediction\", \"label\")\n",
    "evaluator = MulticlassClassificationEvaluator(metricName=\"accuracy\")\n",
    "test_acc = evaluator.evaluate(predictionAndLabels)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MLP Test Accuracy: 0.968\n"
     ]
    }
   ],
   "source": [
    "print('MLP Test Accuracy: {:.3f}'.format(test_acc))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Once finished, end the current context."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}