Nebaditn_HW5_Part2.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{u'Fare': 7.8292, u'Name': u'Kelly, Mr. James', u'Embarked': u'Q', u'Age': 34.5, u'Parch': 0, u'Pclass': 3, u'Sex': u'male', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37c7'), u'PassengerId': 892, u'Ticket': 330911, u'Cabin': u''}\n",
      "{u'Fare': 7, u'Name': u'Wilkes, Mrs. James (Ellen Needs)', u'Embarked': u'S', u'Age': 47, u'Parch': 0, u'Pclass': 3, u'Sex': u'female', u'SibSp': 1, u'_id': ObjectId('5a04a07d2bc46432714b37c8'), u'PassengerId': 893, u'Ticket': 363272, u'Cabin': u''}\n",
      "{u'Fare': 12.2875, u'Name': u'Hirvonen, Mrs. Alexander (Helga E Lindqvist)', u'Embarked': u'S', u'Age': 22, u'Parch': 1, u'Pclass': 3, u'Sex': u'female', u'SibSp': 1, u'_id': ObjectId('5a04a07d2bc46432714b37c9'), u'PassengerId': 896, u'Ticket': 3101298, u'Cabin': u''}\n",
      "{u'Fare': 9.6875, u'Name': u'Myles, Mr. Thomas Francis', u'Embarked': u'Q', u'Age': 62, u'Parch': 0, u'Pclass': 2, u'Sex': u'male', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37ca'), u'PassengerId': 894, u'Ticket': 240276, u'Cabin': u''}\n",
      "{u'Fare': 8.6625, u'Name': u'Wirz, Mr. Albert', u'Embarked': u'S', u'Age': 27, u'Parch': 0, u'Pclass': 3, u'Sex': u'male', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37cb'), u'PassengerId': 895, u'Ticket': 315154, u'Cabin': u''}\n",
      "{u'Fare': 9.225, u'Name': u'Svensson, Mr. Johan Cervin', u'Embarked': u'S', u'Age': 14, u'Parch': 0, u'Pclass': 3, u'Sex': u'male', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37cc'), u'PassengerId': 897, u'Ticket': 7538, u'Cabin': u''}\n",
      "{u'Fare': 7.6292, u'Name': u'Connolly, Miss. Kate', u'Embarked': u'Q', u'Age': 30, u'Parch': 0, u'Pclass': 3, u'Sex': u'female', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37cd'), u'PassengerId': 898, u'Ticket': 330972, u'Cabin': u''}\n",
      "{u'Fare': 29, u'Name': u'Caldwell, Mr. Albert Francis', u'Embarked': u'S', u'Age': 26, u'Parch': 1, u'Pclass': 2, u'Sex': u'male', u'SibSp': 1, u'_id': ObjectId('5a04a07d2bc46432714b37ce'), u'PassengerId': 899, u'Ticket': 248738, u'Cabin': u''}\n",
      "{u'Fare': 7.2292, u'Name': u'Abrahim, Mrs. Joseph (Sophie Halaut Easu)', u'Embarked': u'C', u'Age': 18, u'Parch': 0, u'Pclass': 3, u'Sex': u'female', u'SibSp': 0, u'_id': ObjectId('5a04a07d2bc46432714b37cf'), u'PassengerId': 900, u'Ticket': 2657, u'Cabin': u''}\n",
      "{u'Fare': 24.15, u'Name': u'Davies, Mr. John Samuel', u'Embarked': u'S', u'Age': 21, u'Parch': 0, u'Pclass': 3, u'Sex': u'male', u'SibSp': 2, u'_id': ObjectId('5a04a07d2bc46432714b37d0'), u'PassengerId': 901, u'Ticket': u'A/4 48871', u'Cabin': u''}\n"
     ]
    }
   ],
   "source": [
    "from pymongo import MongoClient\n",
    "client = MongoClient('mongodb://nebadita:Nayak06051989@cluster0-shard-00-00-pak8i.mongodb.net:27017,cluster0-shard-00-01-pak8i.mongodb.net:27017,cluster0-shard-00-02-pak8i.mongodb.net:27017/test?ssl=true&replicaSet=Cluster0-shard-0&authSource=admin')\n",
    "db = client.Titanic\n",
    "collection1 = db.test1\n",
    "collection2 = db.train\n",
    "contents = collection1.find()\n",
    "for line in range(10):\n",
    "    print contents[line]\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 891 entries, 0 to 890\n",
      "Data columns (total 12 columns):\n",
      "Age            891 non-null object\n",
      "Cabin          891 non-null object\n",
      "Embarked       891 non-null object\n",
      "Fare           891 non-null float64\n",
      "Name           891 non-null object\n",
      "Parch          891 non-null object\n",
      "PassengerId    891 non-null int64\n",
      "Pclass         891 non-null int64\n",
      "Sex            891 non-null object\n",
      "SibSp          891 non-null int64\n",
      "Survived       891 non-null int64\n",
      "Ticket         891 non-null object\n",
      "dtypes: float64(1), int64(4), object(7)\n",
      "memory usage: 83.6+ KB\n",
      "  Age Cabin Embarked     Fare  \\\n",
      "0  26              S   7.9250   \n",
      "1  38   C85        C  71.2833   \n",
      "2  35              S   8.0500   \n",
      "3                  Q   8.4583   \n",
      "4  54   E46        S  51.8625   \n",
      "5  35  C123        S  53.1000   \n",
      "6  14              C  30.0708   \n",
      "7   4    G6        S  16.7000   \n",
      "8  58  C103        S  26.5500   \n",
      "9  20              S   8.0500   \n",
      "\n",
      "                                                Name Parch  PassengerId  \\\n",
      "0                             Heikkinen, Miss. Laina     0            3   \n",
      "1  Cumings, Mrs. John Bradley (Florence Briggs Th...     0            2   \n",
      "2                           Allen, Mr. William Henry     0            5   \n",
      "3                                   Moran, Mr. James     0            6   \n",
      "4                            McCarthy, Mr. Timothy J     0            7   \n",
      "5       Futrelle, Mrs. Jacques Heath (Lily May Peel)     0            4   \n",
      "6                Nasser, Mrs. Nicholas (Adele Achem)     0           10   \n",
      "7                    Sandstrom, Miss. Marguerite Rut     1           11   \n",
      "8                           Bonnell, Miss. Elizabeth     0           12   \n",
      "9                     Saundercock, Mr. William Henry     0           13   \n",
      "\n",
      "   Pclass     Sex  SibSp  Survived            Ticket  \n",
      "0       3  female      0         1  STON/O2. 3101282  \n",
      "1       1  female      1         1          PC 17599  \n",
      "2       3    male      0         0            373450  \n",
      "3       3    male      0         0            330877  \n",
      "4       1    male      0         0             17463  \n",
      "5       1  female      1         1            113803  \n",
      "6       2  female      1         1            237736  \n",
      "7       3  female      1         1           PP 9549  \n",
      "8       1  female      0         1            113783  \n",
      "9       3    male      0         0         A/5. 2151  \n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "import pandas\n",
    " \n",
    "spark = SparkSession.builder \\\n",
    "        .appName(\"Titanic\") \\\n",
    "        .getOrCreate()\n",
    "dataFramePandasTrain = pandas.DataFrame(list(collection2.find()))\n",
    "dataFramePandasTest = pandas.DataFrame(list(collection1.find()))\n",
    "dataFramePandasTrain = dataFramePandasTrain.drop('_id', 1)\n",
    "dataFramePandasTest = dataFramePandasTest.drop('_id', 1)\n",
    "dataFramePandasTrain.info()\n",
    "print dataFramePandasTrain.head(10)\n",
    "#print dataFramePandasTest.head(10)\n",
    "\n",
    "#Transforming Pandas Dataframe to Spark DataFrames\n",
    "context = SQLContext(sparkContext = spark.sparkContext, sparkSession = spark)\n",
    "#testRows = dataFramePandasTest.map(lambda p: Row(Age =p[0],Cabin = p[1],Fare = p[2], Name = p[3],Parch =p[4], PassengerId=p[5],Pclass=p[6], Sex=p[7], SibSp=p[8], Survived=p[9], Ticket= p[10]))\n",
    "#sparkTrain = context.createDataFrame(dataFramePandasTrain)\n",
    "#sparkTest = context.createDataFrame(dataFramePandasTest)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 891 entries, 0 to 890\n",
      "Data columns (total 12 columns):\n",
      "Age            714 non-null float64\n",
      "Cabin          891 non-null object\n",
      "Embarked       891 non-null object\n",
      "Fare           891 non-null float64\n",
      "Name           891 non-null object\n",
      "Parch          890 non-null float64\n",
      "PassengerId    891 non-null int64\n",
      "Pclass         891 non-null int64\n",
      "Sex            891 non-null object\n",
      "SibSp          891 non-null int64\n",
      "Survived       891 non-null int64\n",
      "Ticket         891 non-null object\n",
      "dtypes: float64(3), int64(4), object(5)\n",
      "memory usage: 83.6+ KB\n",
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 418 entries, 0 to 417\n",
      "Data columns (total 11 columns):\n",
      "Age            332 non-null float64\n",
      "Cabin          418 non-null object\n",
      "Embarked       418 non-null object\n",
      "Fare           418 non-null object\n",
      "Name           418 non-null object\n",
      "Parch          418 non-null int64\n",
      "PassengerId    418 non-null int64\n",
      "Pclass         418 non-null int64\n",
      "Sex            418 non-null object\n",
      "SibSp          418 non-null int64\n",
      "Ticket         418 non-null object\n",
      "dtypes: float64(1), int64(4), object(6)\n",
      "memory usage: 36.0+ KB\n"
     ]
    }
   ],
   "source": [
    "#Type cast train data from Pandas to dataframe\n",
    "#dataFramePandasTrain.fillna('NaN')\n",
    "dataFramePandasTrain['Age'] = pandas.to_numeric(dataFramePandasTrain['Age'], errors='coerce')\n",
    "dataFramePandasTrain['Parch'] = pandas.to_numeric(dataFramePandasTrain['Parch'], errors='coerce')\n",
    "dataFramePandasTrain['Fare'] = pandas.to_numeric(dataFramePandasTrain['Fare'], errors='ignore')\n",
    "dataFramePandasTrain['Cabin'] = dataFramePandasTrain['Cabin'].astype(basestring).astype(str)\n",
    "dataFramePandasTrain['Embarked'] = dataFramePandasTrain['Embarked'].astype(basestring).astype(str, errors='ignore')\n",
    "dataFramePandasTrain['Sex'] = dataFramePandasTrain['Sex'].astype(basestring).astype(str)\n",
    "dataFramePandasTrain['Ticket'] = dataFramePandasTrain['Ticket'].astype(basestring).astype(str)\n",
    "dataFramePandasTrain.info()\n",
    "train = context.createDataFrame(dataFramePandasTrain)\n",
    "\n",
    "\n",
    "#Type cast test data from Pandas to dataframe\n",
    "\n",
    "dataFramePandasTest['Age'] = pandas.to_numeric(dataFramePandasTest['Age'], errors='coerce')\n",
    "dataFramePandasTest['Cabin'] = dataFramePandasTest['Cabin'].astype(basestring).astype(str)\n",
    "dataFramePandasTest['Embarked'] = dataFramePandasTest['Embarked'].astype(basestring).astype(str)\n",
    "dataFramePandasTest['Age'] = pandas.to_numeric(dataFramePandasTest['Age'])\n",
    "dataFramePandasTest['Sex'] = dataFramePandasTest['Sex'].astype(basestring).astype(str)\n",
    "dataFramePandasTest['Ticket'] = dataFramePandasTest['Ticket'].astype(basestring).astype(str)\n",
    "dataFramePandasTest.info()\n",
    "\n",
    "test = context.createDataFrame(dataFramePandasTrain)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "| Age|Cabin|Embarked|   Fare|                Name|Parch|PassengerId|Pclass|   Sex|SibSp|label|          Ticket|modifiedAge|modifiedFare|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "|26.0|     |       S|  7.925|Heikkinen, Miss. ...|  0.0|          3|     3|female|    0|    1|STON/O2. 3101282|       26.0|       7.925|\n",
      "|38.0|  C85|       C|71.2833|Cumings, Mrs. Joh...|  0.0|          2|     1|female|    1|    1|        PC 17599|       38.0|     71.2833|\n",
      "|35.0|     |       S|   8.05|Allen, Mr. Willia...|  0.0|          5|     3|  male|    0|    0|          373450|       35.0|        8.05|\n",
      "| 0.0|     |       Q| 8.4583|    Moran, Mr. James|  0.0|          6|     3|  male|    0|    0|          330877|        0.0|      8.4583|\n",
      "|54.0|  E46|       S|51.8625|McCarthy, Mr. Tim...|  0.0|          7|     1|  male|    0|    0|           17463|       54.0|     51.8625|\n",
      "|35.0| C123|       S|   53.1|Futrelle, Mrs. Ja...|  0.0|          4|     1|female|    1|    1|          113803|       35.0|        53.1|\n",
      "|14.0|     |       C|30.0708|Nasser, Mrs. Nich...|  0.0|         10|     2|female|    1|    1|          237736|       14.0|     30.0708|\n",
      "| 4.0|   G6|       S|   16.7|Sandstrom, Miss. ...|  1.0|         11|     3|female|    1|    1|         PP 9549|        4.0|        16.7|\n",
      "|58.0| C103|       S|  26.55|Bonnell, Miss. El...|  0.0|         12|     1|female|    0|    1|          113783|       58.0|       26.55|\n",
      "|20.0|     |       S|   8.05|Saundercock, Mr. ...|  0.0|         13|     3|  male|    0|    0|       A/5. 2151|       20.0|        8.05|\n",
      "| 2.0|     |       S| 21.075|Palsson, Master. ...|  1.0|          8|     3|  male|    3|    0|          349909|        2.0|      21.075|\n",
      "|27.0|     |       S|11.1333|Johnson, Mrs. Osc...|  2.0|          9|     3|female|    0|    1|          347742|       27.0|     11.1333|\n",
      "|39.0|     |       S| 31.275|Andersson, Mr. An...|  5.0|         14|     3|  male|    1|    0|          347082|       39.0|      31.275|\n",
      "|14.0|     |       S| 7.8542|Vestrom, Miss. Hu...|  0.0|         15|     3|female|    0|    0|          350406|       14.0|      7.8542|\n",
      "|55.0|     |       S|   16.0|Hewlett, Mrs. (Ma...|  0.0|         16|     2|female|    0|    1|          248706|       55.0|        16.0|\n",
      "| 2.0|     |       Q| 29.125|Rice, Master. Eugene|  1.0|         17|     3|  male|    4|    0|          382652|        2.0|      29.125|\n",
      "| 0.0|     |       S|   13.0|Williams, Mr. Cha...|  0.0|         18|     2|  male|    0|    1|          244373|        0.0|        13.0|\n",
      "|31.0|     |       S|   18.0|Vander Planke, Mr...|  0.0|         19|     3|female|    1|    0|          345763|       31.0|        18.0|\n",
      "|35.0|     |       S|   26.0|Fynney, Mr. Joseph J|  0.0|         21|     2|  male|    0|    0|          239865|       35.0|        26.0|\n",
      "| 0.0|     |       C|  7.225|Masselmani, Mrs. ...|  NaN|         20|     3|female|    0|    1|            2649|        0.0|       7.225|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "| Age|Cabin|Embarked|   Fare|                Name|Parch|PassengerId|Pclass|   Sex|SibSp|label|          Ticket|modifiedAge|modifiedFare|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "|26.0|     |       S|  7.925|Heikkinen, Miss. ...|  0.0|          3|     3|female|    0|    1|STON/O2. 3101282|       26.0|       7.925|\n",
      "|38.0|  C85|       C|71.2833|Cumings, Mrs. Joh...|  0.0|          2|     1|female|    1|    1|        PC 17599|       38.0|     71.2833|\n",
      "|35.0|     |       S|   8.05|Allen, Mr. Willia...|  0.0|          5|     3|  male|    0|    0|          373450|       35.0|        8.05|\n",
      "| 0.0|     |       Q| 8.4583|    Moran, Mr. James|  0.0|          6|     3|  male|    0|    0|          330877|        0.0|      8.4583|\n",
      "|54.0|  E46|       S|51.8625|McCarthy, Mr. Tim...|  0.0|          7|     1|  male|    0|    0|           17463|       54.0|     51.8625|\n",
      "|35.0| C123|       S|   53.1|Futrelle, Mrs. Ja...|  0.0|          4|     1|female|    1|    1|          113803|       35.0|        53.1|\n",
      "|14.0|     |       C|30.0708|Nasser, Mrs. Nich...|  0.0|         10|     2|female|    1|    1|          237736|       14.0|     30.0708|\n",
      "| 4.0|   G6|       S|   16.7|Sandstrom, Miss. ...|  1.0|         11|     3|female|    1|    1|         PP 9549|        4.0|        16.7|\n",
      "|58.0| C103|       S|  26.55|Bonnell, Miss. El...|  0.0|         12|     1|female|    0|    1|          113783|       58.0|       26.55|\n",
      "|20.0|     |       S|   8.05|Saundercock, Mr. ...|  0.0|         13|     3|  male|    0|    0|       A/5. 2151|       20.0|        8.05|\n",
      "| 2.0|     |       S| 21.075|Palsson, Master. ...|  1.0|          8|     3|  male|    3|    0|          349909|        2.0|      21.075|\n",
      "|27.0|     |       S|11.1333|Johnson, Mrs. Osc...|  2.0|          9|     3|female|    0|    1|          347742|       27.0|     11.1333|\n",
      "|39.0|     |       S| 31.275|Andersson, Mr. An...|  5.0|         14|     3|  male|    1|    0|          347082|       39.0|      31.275|\n",
      "|14.0|     |       S| 7.8542|Vestrom, Miss. Hu...|  0.0|         15|     3|female|    0|    0|          350406|       14.0|      7.8542|\n",
      "|55.0|     |       S|   16.0|Hewlett, Mrs. (Ma...|  0.0|         16|     2|female|    0|    1|          248706|       55.0|        16.0|\n",
      "| 2.0|     |       Q| 29.125|Rice, Master. Eugene|  1.0|         17|     3|  male|    4|    0|          382652|        2.0|      29.125|\n",
      "| 0.0|     |       S|   13.0|Williams, Mr. Cha...|  0.0|         18|     2|  male|    0|    1|          244373|        0.0|        13.0|\n",
      "|31.0|     |       S|   18.0|Vander Planke, Mr...|  0.0|         19|     3|female|    1|    0|          345763|       31.0|        18.0|\n",
      "|35.0|     |       S|   26.0|Fynney, Mr. Joseph J|  0.0|         21|     2|  male|    0|    0|          239865|       35.0|        26.0|\n",
      "| 0.0|     |       C|  7.225|Masselmani, Mrs. ...|  NaN|         20|     3|female|    0|    1|            2649|        0.0|       7.225|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "test = test.fillna({'Age':'0'})\n",
    "train = train.fillna({'Age':'0'})\n",
    "train = train.fillna({'Embarked':'S'})\n",
    "test = test.fillna({'Embarked':'0'})\n",
    "print test.show()\n",
    "print train.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------------------+\n",
      "|Pclass|          avg(Age)|\n",
      "+------+------------------+\n",
      "|     1| 32.92324074074074|\n",
      "|     3|18.177026476578412|\n",
      "|     2| 28.09146739130435|\n",
      "+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Since from the above step we realized that the age column has null values, we will try to gauge the average age of people belonging to a particular class.\n",
    "\n",
    "exploreAge = train.where(train[\"Age\"].isNotNull()).groupBy('Pclass').avg('Age')\n",
    "exploreAge.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "|Age|Cabin|Embarked|Fare|Name|Parch|PassengerId|Pclass|Sex|SibSp|label|Ticket|modifiedAge|modifiedFare|\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import Imputer\n",
    "#Now we will check for null values in the Age column\n",
    "NullAgeTrain = train.where(train[\"Age\"].isNull())\n",
    "#display(NullFaresTrain)\n",
    "\n",
    "#Check Null fares for Test\n",
    "NullAgeTest = test.where(test[\"Age\"].isNull())\n",
    "#display(NullFaresTest)\n",
    "\n",
    "#From the above queries we will realize that there is one Null value for Test data. That is for passengerID = 1044\n",
    "\n",
    "#Now we will replace this Null value with the mean value of the column. To do so we will use the Imputer transformer\n",
    "imputer= Imputer(inputCols=[\"Age\"],outputCols=[\"modifiedAge\"]).fit(test)\n",
    "test = imputer.transform(test)\n",
    "\n",
    "#Do it for train too , to maintain the sync of columns in test and train\n",
    "train = imputer.transform(train)\n",
    "\n",
    "#Check Null fares for Test after running Imputer\n",
    "NullAgeTest = test.where(test[\"Age\"].isNull())\n",
    "NullAgeTest.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "|Age|Cabin|Embarked|Fare|Name|Parch|PassengerId|Pclass|Sex|SibSp|label|Ticket|modifiedAge|modifiedFare|\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#I am including this step in my second iteration. I discovered that my Vector Assembler was erroring out in the pipeline because there were null categorical values in the\n",
    "#Error received: Caused by: org.apache.spark.SparkException: Values to assemble cannot be null.\n",
    "#Embarked column for for Passenger ID 62 and 830.\n",
    "\n",
    "exploreEmbarkNull = train.where(train[\"Embarked\"].isNull())\n",
    "exploreEmbarkNull.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|Embarked|count|\n",
      "+--------+-----+\n",
      "+--------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# After discovering the null values, I want to replace them with the highest frequency of Embarked class occuring in the train data.\n",
    "#From the below query, I realized tha\n",
    "#Embarked   #Count\n",
    "#Q          77\n",
    "#C          168\n",
    "#S          644\n",
    "\n",
    "#So, will replace the two null embarked values [Passenger ID: 62,830] with Embarked Values \"S\" as frequency of null values\n",
    "\n",
    "exploreEmbark = train.where(train[\"Embarked\"].isNotNull()).groupBy('Embarked').count()\n",
    "exploreEmbark.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "|Age|Cabin|Embarked|Fare|Name|Parch|PassengerId|Pclass|Sex|SibSp|label|Ticket|modifiedAge|modifiedFare|\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "##Replae the Embarked valuees to 'S' for passenger 62 and 830\n",
    "train = train.na.fill({'Embarked':'S'})\n",
    "test = test.na.fill({'Embarked':'S'})\n",
    "\n",
    "#Check the values\n",
    "embarkedNullRemoved = train.where(train[\"Embarked\"].isNull())\n",
    "embarkedNullRemoved.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "|Age|Cabin|Embarked|Fare|Name|Parch|PassengerId|Pclass|Sex|SibSp|label|Ticket|modifiedAge|modifiedFare|\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "+---+-----+--------+----+----+-----+-----------+------+---+-----+-----+------+-----------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import Imputer\n",
    "#Now we will check for null values in the Fare column\n",
    "NullFaresTrain = train.where(train[\"Fare\"].isNull())\n",
    "#display(NullFaresTrain)\n",
    "\n",
    "#Check Null fares for Test\n",
    "NullFaresTest = test.where(test[\"Fare\"].isNull())\n",
    "#display(NullFaresTest)\n",
    "\n",
    "#From the above queries we will realize that there is one Null value for Test data. That is for passengerID = 1044\n",
    "\n",
    "#Now we will replace this Null value with the mean value of the column. To do so we will use the Imputer transformer\n",
    "imputer= Imputer(inputCols=[\"Fare\"],outputCols=[\"modifiedFare\"]).fit(test)\n",
    "test = imputer.transform(test)\n",
    "\n",
    "#Do it for train too , to maintain the sync of columns in test and train\n",
    "train = imputer.transform(train)\n",
    "\n",
    "#Check Null fares for Test after running Imputer\n",
    "NullRemovedFaresTest = test.where(test[\"Fare\"].isNull())\n",
    "NullRemovedFaresTest.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "| Age|Cabin|Embarked|   Fare|                Name|Parch|PassengerId|Pclass|   Sex|SibSp|label|          Ticket|modifiedAge|modifiedFare|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "|26.0|     |       S|  7.925|Heikkinen, Miss. ...|  0.0|          3|     3|female|    0|    1|STON/O2. 3101282|       26.0|       7.925|\n",
      "|38.0|  C85|       C|71.2833|Cumings, Mrs. Joh...|  0.0|          2|     1|female|    1|    1|        PC 17599|       38.0|     71.2833|\n",
      "|35.0|     |       S|   8.05|Allen, Mr. Willia...|  0.0|          5|     3|  male|    0|    0|          373450|       35.0|        8.05|\n",
      "| 0.0|     |       Q| 8.4583|    Moran, Mr. James|  0.0|          6|     3|  male|    0|    0|          330877|        0.0|      8.4583|\n",
      "|54.0|  E46|       S|51.8625|McCarthy, Mr. Tim...|  0.0|          7|     1|  male|    0|    0|           17463|       54.0|     51.8625|\n",
      "|35.0| C123|       S|   53.1|Futrelle, Mrs. Ja...|  0.0|          4|     1|female|    1|    1|          113803|       35.0|        53.1|\n",
      "|14.0|     |       C|30.0708|Nasser, Mrs. Nich...|  0.0|         10|     2|female|    1|    1|          237736|       14.0|     30.0708|\n",
      "| 4.0|   G6|       S|   16.7|Sandstrom, Miss. ...|  1.0|         11|     3|female|    1|    1|         PP 9549|        4.0|        16.7|\n",
      "|58.0| C103|       S|  26.55|Bonnell, Miss. El...|  0.0|         12|     1|female|    0|    1|          113783|       58.0|       26.55|\n",
      "|20.0|     |       S|   8.05|Saundercock, Mr. ...|  0.0|         13|     3|  male|    0|    0|       A/5. 2151|       20.0|        8.05|\n",
      "| 2.0|     |       S| 21.075|Palsson, Master. ...|  1.0|          8|     3|  male|    3|    0|          349909|        2.0|      21.075|\n",
      "|27.0|     |       S|11.1333|Johnson, Mrs. Osc...|  2.0|          9|     3|female|    0|    1|          347742|       27.0|     11.1333|\n",
      "|39.0|     |       S| 31.275|Andersson, Mr. An...|  5.0|         14|     3|  male|    1|    0|          347082|       39.0|      31.275|\n",
      "|14.0|     |       S| 7.8542|Vestrom, Miss. Hu...|  0.0|         15|     3|female|    0|    0|          350406|       14.0|      7.8542|\n",
      "|55.0|     |       S|   16.0|Hewlett, Mrs. (Ma...|  0.0|         16|     2|female|    0|    1|          248706|       55.0|        16.0|\n",
      "| 2.0|     |       Q| 29.125|Rice, Master. Eugene|  1.0|         17|     3|  male|    4|    0|          382652|        2.0|      29.125|\n",
      "| 0.0|     |       S|   13.0|Williams, Mr. Cha...|  0.0|         18|     2|  male|    0|    1|          244373|        0.0|        13.0|\n",
      "|31.0|     |       S|   18.0|Vander Planke, Mr...|  0.0|         19|     3|female|    1|    0|          345763|       31.0|        18.0|\n",
      "|35.0|     |       S|   26.0|Fynney, Mr. Joseph J|  0.0|         21|     2|  male|    0|    0|          239865|       35.0|        26.0|\n",
      "| 0.0|     |       C|  7.225|Masselmani, Mrs. ...|  NaN|         20|     3|female|    0|    1|            2649|        0.0|       7.225|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "| Age|Cabin|Embarked|   Fare|                Name|Parch|PassengerId|Pclass|   Sex|SibSp|label|          Ticket|modifiedAge|modifiedFare|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "|26.0|     |       S|  7.925|Heikkinen, Miss. ...|  0.0|          3|     3|female|    0|    1|STON/O2. 3101282|       26.0|       7.925|\n",
      "|38.0|  C85|       C|71.2833|Cumings, Mrs. Joh...|  0.0|          2|     1|female|    1|    1|        PC 17599|       38.0|     71.2833|\n",
      "|35.0|     |       S|   8.05|Allen, Mr. Willia...|  0.0|          5|     3|  male|    0|    0|          373450|       35.0|        8.05|\n",
      "| 0.0|     |       Q| 8.4583|    Moran, Mr. James|  0.0|          6|     3|  male|    0|    0|          330877|        0.0|      8.4583|\n",
      "|54.0|  E46|       S|51.8625|McCarthy, Mr. Tim...|  0.0|          7|     1|  male|    0|    0|           17463|       54.0|     51.8625|\n",
      "|35.0| C123|       S|   53.1|Futrelle, Mrs. Ja...|  0.0|          4|     1|female|    1|    1|          113803|       35.0|        53.1|\n",
      "|14.0|     |       C|30.0708|Nasser, Mrs. Nich...|  0.0|         10|     2|female|    1|    1|          237736|       14.0|     30.0708|\n",
      "| 4.0|   G6|       S|   16.7|Sandstrom, Miss. ...|  1.0|         11|     3|female|    1|    1|         PP 9549|        4.0|        16.7|\n",
      "|58.0| C103|       S|  26.55|Bonnell, Miss. El...|  0.0|         12|     1|female|    0|    1|          113783|       58.0|       26.55|\n",
      "|20.0|     |       S|   8.05|Saundercock, Mr. ...|  0.0|         13|     3|  male|    0|    0|       A/5. 2151|       20.0|        8.05|\n",
      "| 2.0|     |       S| 21.075|Palsson, Master. ...|  1.0|          8|     3|  male|    3|    0|          349909|        2.0|      21.075|\n",
      "|27.0|     |       S|11.1333|Johnson, Mrs. Osc...|  2.0|          9|     3|female|    0|    1|          347742|       27.0|     11.1333|\n",
      "|39.0|     |       S| 31.275|Andersson, Mr. An...|  5.0|         14|     3|  male|    1|    0|          347082|       39.0|      31.275|\n",
      "|14.0|     |       S| 7.8542|Vestrom, Miss. Hu...|  0.0|         15|     3|female|    0|    0|          350406|       14.0|      7.8542|\n",
      "|55.0|     |       S|   16.0|Hewlett, Mrs. (Ma...|  0.0|         16|     2|female|    0|    1|          248706|       55.0|        16.0|\n",
      "| 2.0|     |       Q| 29.125|Rice, Master. Eugene|  1.0|         17|     3|  male|    4|    0|          382652|        2.0|      29.125|\n",
      "| 0.0|     |       S|   13.0|Williams, Mr. Cha...|  0.0|         18|     2|  male|    0|    1|          244373|        0.0|        13.0|\n",
      "|31.0|     |       S|   18.0|Vander Planke, Mr...|  0.0|         19|     3|female|    1|    0|          345763|       31.0|        18.0|\n",
      "|35.0|     |       S|   26.0|Fynney, Mr. Joseph J|  0.0|         21|     2|  male|    0|    0|          239865|       35.0|        26.0|\n",
      "| 0.0|     |       C|  7.225|Masselmani, Mrs. ...|  NaN|         20|     3|female|    0|    1|            2649|        0.0|       7.225|\n",
      "+----+-----+--------+-------+--------------------+-----+-----------+------+------+-----+-----+----------------+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\n",
    "from pyspark.ml.feature import *\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "# Rename survived column to \"label\"\n",
    "train = train.withColumnRenamed(\"Survived\", \"label\")\n",
    "test = test.withColumnRenamed(\"Survived\", \"label\")\n",
    "\n",
    "train.show()\n",
    "test.show()\n",
    "\n",
    "\n",
    "\n",
    "#List of Categorical Variables\n",
    "#1) Survived: Label 1 or 0\n",
    "#2) Pclass: Categorical (1,2,3)\n",
    "#3) Sex: Categorical (Male,Female)\n",
    "#4) Age: Continuous\n",
    "#5) SibSp: Categorical(0,1,2,...8) ---Discrete\n",
    "#6) Parch: Categorical(0,1,2,3...9)----Discrete\n",
    "#7) Fare: Continuous\n",
    "#8) Embarked: Categorical(C,Q,S)\n",
    "\n",
    "\n",
    "#Indexed all the labels, to add the metadata\n",
    "genderIndex = StringIndexer(inputCol = \"Sex\", outputCol=\"indexedSex\" ).setHandleInvalid(\"skip\")\n",
    "embarkIndex = StringIndexer(inputCol=\"Embarked\", outputCol=\"indexedEmbarked\").setHandleInvalid(\"skip\")\n",
    "\n",
    "#After multiple iterations I realized that we cannot use OHE on features with different categories in the test and the train data set. Because the Vector assembler will error out in the Pipeline.tranform() step due to mismatch in the vector size. Hence we are only using OHE on the gender and Embarked category as their number of categories is constant in the test and in the train data set.\n",
    "\n",
    "#Creating One hot encoding on indexed columns to produce sparse vectors\n",
    "#surviveEncode = OneHotEncoder(inputCol=\"Survived\",outputCol=\"survivedVector\")\n",
    "genderEncode = OneHotEncoder(inputCol=genderIndex.getOutputCol(),outputCol=\"SexVector\")\n",
    "embarkEncode = OneHotEncoder(inputCol=embarkIndex.getOutputCol(),outputCol=\"EmbarkedVector\")\n",
    "pClassEncode = OneHotEncoder(inputCol=\"Pclass\", outputCol=\"PclassVector\")\n",
    "SibSpEncode = OneHotEncoder(inputCol=\"SibSp\",outputCol=\"SibSpVector\")\n",
    "ParchEncode = OneHotEncoder(inputCol=\"Parch\",outputCol=\"ParchVector\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Age: double, Cabin: string, Embarked: string, Fare: double, Name: string, Parch: double, PassengerId: bigint, Pclass: bigint, Sex: string, SibSp: bigint, label: bigint, Ticket: string, modifiedAge: double, modifiedFare: double]"
      ]
     },
     "execution_count": 150,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Transform all features into a vector using VectorAssembler\n",
    "from pyspark.ml.linalg import Vectors\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "numericalCols = [\"modifiedAge\",\"modifiedFare\"]\n",
    "#categoricalCols = [\"Pclass\",\"Sex\",\"SibSp\",\"Parch\",\"Embarked\"]\n",
    "#categoricalCols = [\"Sex\",\"Embarked\"]\n",
    "categoricalCols = [\"Sex\"]\n",
    "#otherCategoricalCols = [\"Pclass\",\"SibSp\",\"Parch\"]\n",
    "otherCategoricalCols = [\"Pclass\",\"SibSp\"]\n",
    "#assemblerInputs = map(lambda c: c + \"Vector\", categoricalCols) + numericalCols\n",
    "assemblerInputs = map(lambda c: c + \"Vector\", categoricalCols) + numericalCols +otherCategoricalCols\n",
    "assembler = VectorAssembler(inputCols=assemblerInputs, outputCol=\"features\")\n",
    "test.fillna(0)\n",
    "train.fillna(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#First we will fit a Logistic Regression model for this Binary Classification problem. That is to predict whether a person survived or not (0,1).\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "(trainingData, testData) = train.randomSplit([0.7, 0.3], seed = 100)\n",
    "\n",
    "# Create initial LogisticRegression model\n",
    "lr = LogisticRegression(labelCol=\"label\", featuresCol=\"features\", maxIter=10)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Age: double (nullable = true)\n",
      " |-- Cabin: string (nullable = true)\n",
      " |-- Embarked: string (nullable = false)\n",
      " |-- Fare: double (nullable = true)\n",
      " |-- Name: string (nullable = true)\n",
      " |-- Parch: double (nullable = true)\n",
      " |-- PassengerId: long (nullable = true)\n",
      " |-- Pclass: long (nullable = true)\n",
      " |-- Sex: string (nullable = true)\n",
      " |-- SibSp: long (nullable = true)\n",
      " |-- label: long (nullable = true)\n",
      " |-- Ticket: string (nullable = true)\n",
      " |-- modifiedAge: double (nullable = true)\n",
      " |-- modifiedFare: double (nullable = true)\n",
      " |-- indexedSex: double (nullable = true)\n",
      " |-- indexedEmbarked: double (nullable = true)\n",
      " |-- SexVector: vector (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- rawPrediction: vector (nullable = true)\n",
      " |-- probability: vector (nullable = true)\n",
      " |-- prediction: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Execute the model\n",
    "from pyspark.ml import Pipeline\n",
    "pipeline = Pipeline(stages=[genderIndex,embarkIndex,genderEncode,assembler,lr])\n",
    "#pipeline = Pipeline(stages=[genderIndex,embarkIndex,genderEncode,embarkEncode,pClassEncode,SibSpEncode,ParchEncode])\n",
    "model = pipeline.fit(trainingData)\n",
    "predictions = model.transform(testData)\n",
    "\n",
    "#Check the schema\n",
    "predictions.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 155,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8532255065997397"
      ]
     },
     "execution_count": 155,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Now we will evalute our model\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "evaluator = BinaryClassificationEvaluator(rawPredictionCol=\"rawPrediction\")\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 156,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Now we will try tuning the model with the ParamGridBuilder and the CrossValidator.\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "# Create ParamGrid for Cross Validation\n",
    "paramGrid = (ParamGridBuilder()\n",
    "             .addGrid(lr.regParam, [0.01, 0.5, 2.0])\n",
    "             .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])\n",
    "             .addGrid(lr.maxIter, [1, 5, 10])\n",
    "             .build())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "# Create 5-fold CrossValidator\n",
    "cv = CrossValidator(estimator=lr, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5)\n",
    "\n",
    "pipelineforEval = Pipeline(stages=[genderIndex,embarkIndex,genderEncode,embarkEncode,assembler]).fit(train)\n",
    "cvData = pipelineforEval.transform(train)\n",
    "# Run cross validations\n",
    "cvModel = cv.fit(cvData.select(\"label\",\"features\"))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Use test set here so we can measure the accuracy of our model on new data\n",
    "\n",
    "cvData = pipelineforEval.transform(testData)\n",
    "predictions = cvModel.transform(cvData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8649996901530645"
      ]
     },
     "execution_count": 159,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# cvModel uses the best model found from the Cross Validation\n",
    "# Evaluate best model\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 160,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+\n",
      "|PassengerId|prediction|\n",
      "+-----------+----------+\n",
      "|          3|       1.0|\n",
      "|          2|       1.0|\n",
      "|          5|       0.0|\n",
      "|          6|       0.0|\n",
      "|          7|       0.0|\n",
      "|          4|       1.0|\n",
      "|         10|       1.0|\n",
      "|         11|       1.0|\n",
      "|         12|       1.0|\n",
      "|         13|       0.0|\n",
      "|          8|       0.0|\n",
      "|          9|       1.0|\n",
      "|         14|       0.0|\n",
      "|         15|       1.0|\n",
      "|         16|       1.0|\n",
      "|         17|       0.0|\n",
      "|         18|       0.0|\n",
      "|         19|       1.0|\n",
      "|         21|       0.0|\n",
      "|         20|       1.0|\n",
      "+-----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Download the CSV for test Data\n",
    "cvData = pipelineforEval.transform(test)\n",
    "predictions = cvModel.transform(cvData)\n",
    "#predictions.printSchema()\n",
    "predictions.select(\"PassengerId\",\"prediction\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 162,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Now we will train our model using a random Forest Classifier\n",
    "#Train a random classifier\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "rf = RandomForestClassifier(labelCol=\"label\",featuresCol=\"features\")\n",
    "rfPipeline = Pipeline(stages=[genderIndex,embarkIndex,genderEncode,embarkEncode,assembler,rf])\n",
    "\n",
    "rfModel = rfPipeline.fit(trainingData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Make predictions on test data using the Transformer.transform() method.\n",
    "predictions = rfModel.transform(testData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 164,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8944041643428148"
      ]
     },
     "execution_count": 164,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "# Evaluate model\n",
    "evaluator = BinaryClassificationEvaluator()\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 165,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create ParamGrid for Cross Validation\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "paramGrid = (ParamGridBuilder()\n",
    "             .addGrid(rf.maxDepth, [2, 4, 6])\n",
    "             .addGrid(rf.maxBins, [20, 60])\n",
    "             .addGrid(rf.numTrees, [5, 20])\n",
    "             .build())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 166,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create 5-fold CrossValidator\n",
    "rfCV = CrossValidator(estimator=rf, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5)\n",
    "\n",
    "cvData = pipelineforEval.transform(trainingData)\n",
    "# Run cross validations\n",
    "cvModelRF = rfCV.fit(cvData.select(\"label\",\"features\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 167,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8944041643428147"
      ]
     },
     "execution_count": 167,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Use test set here so we can measure the accuracy of our model on new data\n",
    "cvDataTest = pipelineforEval.transform(testData)\n",
    "\n",
    "rfPredictions = cvModelRF.transform(cvDataTest.select(\"label\",\"features\"))\n",
    "evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 168,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Select the best Model\n",
    "bestModel = cvModelRF.bestModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 169,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Make predictions for the entire test data set\n",
    "cvDataTest = pipelineforEval.transform(test)\n",
    "finalPredictions = bestModel.transform(cvDataTest)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 170,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+\n",
      "|passengerID|prediction|\n",
      "+-----------+----------+\n",
      "|          3|       1.0|\n",
      "|          2|       1.0|\n",
      "|          5|       0.0|\n",
      "|          6|       0.0|\n",
      "|          7|       0.0|\n",
      "|          4|       1.0|\n",
      "|         10|       1.0|\n",
      "|         11|       1.0|\n",
      "|         12|       1.0|\n",
      "|         13|       0.0|\n",
      "|          8|       0.0|\n",
      "|          9|       1.0|\n",
      "|         14|       0.0|\n",
      "|         15|       1.0|\n",
      "|         16|       1.0|\n",
      "|         17|       0.0|\n",
      "|         18|       0.0|\n",
      "|         19|       0.0|\n",
      "|         21|       0.0|\n",
      "|         20|       1.0|\n",
      "+-----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "finalPredictions.select(\"passengerID\",\"prediction\").show()\n",
    "finalPredictions.select(\"passengerID\",\"prediction\").count()\n",
    "\n",
    "result = finalPredictions.select(\"passengerID\",\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 172,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pymongo.results.InsertManyResult at 0x7f412ef0b4b0>"
      ]
     },
     "execution_count": 172,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "resultPd = result.toPandas()\n",
    "newCollection = 'TitanicResult'\n",
    "newColToSave = db[newCollection]\n",
    "jsonData = json.loads(resultPd.to_json(orient='records'))\n",
    "\n",
    "newColToSave.insert_many(jsonData)"
   ]
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}