titanic.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Populating the interactive namespace from numpy and matplotlib\n"
     ]
    }
   ],
   "source": [
    "%pylab inline\n",
    "\n",
    "import numpy as np\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import functions\n",
    "from pyspark.ml.feature import VectorAssembler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    " def prepare_data(data):\n",
    "    ## Fill missing Age values with the average age per class\n",
    "\n",
    "    explore_age_df = data.orderBy('Age', ascending=True)\n",
    "\n",
    "    avg_age_df = explore_age_df.where(explore_age_df['Age'].isNotNull()).groupBy('Pclass').avg('Age')\n",
    "    avg_age_df = avg_age_df.select('Pclass', avg_age_df['avg(Age)'].alias('Age'))\n",
    "\n",
    "    avg_age_list = avg_age_df.collect()\n",
    "\n",
    "    # Replace null values with the average age values from our passenger class list\n",
    "    data_with_age_df = (data\n",
    "                         .select('*', \n",
    "                                 when(data['Age'].isNull() & (data['Pclass'] == 1), \n",
    "                                      avg_age_list[0].Age)\n",
    "                                 .otherwise(when(data['Age'].isNull() & (data['Pclass'] == 2), \n",
    "                                                 avg_age_list[1].Age)\n",
    "                                            .otherwise(when(data['Age'].isNull() & (data['Pclass'] == 3), \n",
    "                                                            avg_age_list[2].Age)\n",
    "                                                       .otherwise(col('Age')))).alias('FilledAge')))\n",
    "\n",
    "    # Replace the Age column values with those from our FilledAge column and then drop FilledAge.\n",
    "    data_with_age_df = data_with_age_df.withColumn('Age', data_with_age_df['FilledAge']).drop('FilledAge')\n",
    "\n",
    "    ## Index Sex\n",
    "\n",
    "    def sex_to_int(sex):\n",
    "      if(sex.lower() == 'male'):\n",
    "        return 0\n",
    "      else:\n",
    "        return 1\n",
    "    sex_classify = functions.udf(sex_to_int, IntegerType())\n",
    "\n",
    "    sex_int_df = data_with_age_df.select('*', sex_classify(data_with_age_df['Sex']).alias('IntSex'))\n",
    "    data_sex_indexed_df = sex_int_df.withColumn('Sex', sex_int_df['IntSex']).drop('IntSex').cache()\n",
    "\n",
    "    data_sex_indexed_df\n",
    "\n",
    "    ## Index Cabin\n",
    "\n",
    "    def cabin_to_int(cabin):\n",
    "        if cabin:\n",
    "            return ord(cabin[0])-ord('A')+1 #A:1; B:2; C:3; D:4; None:0\n",
    "        else:\n",
    "            return 0\n",
    "    cabin_classify = functions.udf(cabin_to_int, IntegerType())\n",
    "\n",
    "    cabin_int_df = data_sex_indexed_df.select('*', cabin_classify(data_sex_indexed_df['Cabin']).alias('IntCabin'))\n",
    "    data_cabin_indexed_df = cabin_int_df.withColumn('Cabin', cabin_int_df['IntCabin']).drop('IntCabin').cache()\n",
    "\n",
    "    data_cabin_indexed_df\n",
    "\n",
    "    ##  Index Embarked\n",
    "\n",
    "    def embarked_to_int(embarked):\n",
    "        if embarked == 'C':   #Cherbourg; \n",
    "            return 1 #TBD\n",
    "        elif embarked == 'Q': #Queenstown; \n",
    "            return 2\n",
    "        elif embarked == 'S': #Southampton)\n",
    "            return 3    \n",
    "        else:\n",
    "            return 0\n",
    "    embarked_classify = functions.udf(embarked_to_int, IntegerType())\n",
    "\n",
    "    embarked_int_df = data_cabin_indexed_df.select('*', embarked_classify(data_sex_indexed_df['Embarked']).alias('IntEmbarked'))\n",
    "    data_embarked_indexed_df = embarked_int_df.withColumn('Embarked', embarked_int_df['IntEmbarked']).drop('IntEmbarked').cache()\n",
    "\n",
    "    return data_embarked_indexed_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Intitialize Train data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+------+--------------------+---+----+-----+-----+---------+----+-----+--------+\n",
      "|PassengerId|Survived|Pclass|                Name|Sex| Age|SibSp|Parch|   Ticket|Fare|Cabin|Embarked|\n",
      "+-----------+--------+------+--------------------+---+----+-----+-----+---------+----+-----+--------+\n",
      "|          1|       0|     3|Braund, Mr. Owen ...|  0|22.0|    1|    0|A/5 21171|7.25|    0|       3|\n",
      "+-----------+--------+------+--------------------+---+----+-----+-----+---------+----+-----+--------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train_data = (sqlContext\n",
    "                 .read\n",
    "                 .format('csv')\n",
    "                 .options(header='true', inferSchema='true')\n",
    "                 .load('./data/train.csv'))\n",
    "train_data.cache()\n",
    "train_data = prepare_data(train_data)\n",
    "train_data.show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Assemble feature vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+-------------------------------+\n",
      "|PassengerId|label|features                       |\n",
      "+-----------+-----+-------------------------------+\n",
      "|1          |0    |[3.0,0.0,22.0,1.0,0.0,7.25,3.0]|\n",
      "+-----------+-----+-------------------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "assembler = VectorAssembler(\n",
    "    inputCols=['Pclass','Sex','Age','SibSp','Parch','Fare','Embarked'], outputCol='features'\n",
    ")\n",
    "\n",
    "train_data = assembler.transform(train_data).select('PassengerId',col('Survived').alias('label'),'features')\n",
    "train_data.show(1,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(DataFrame[PassengerId: int, label: int, features: vector],\n",
       " DataFrame[PassengerId: int, label: int, features: vector])"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "splits = train_data.randomSplit([0.8, 0.2])\n",
    "train_train = splits[0].cache() #caching brings significant ~30% perofromance improvement to fitting\n",
    "train_test = splits[1].cache()\n",
    "train_train, train_test"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Benchmarking various classifiers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NaiveBayes                    \t0.699453551913\n",
      "LogisticRegression            \t0.803278688525\n",
      "RandomForestClassifier        \t0.819672131148\n",
      "MultilayerPerceptronClassifier\t0.79781420765\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.classification import *\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "\n",
    "for classifier in (NaiveBayes, LogisticRegression, RandomForestClassifier, MultilayerPerceptronClassifier): # '[]' won't work\n",
    "    \n",
    "    if classifier != MultilayerPerceptronClassifier:\n",
    "        model = classifier()\n",
    "    else:\n",
    "        #Number of inputs = the size of feature vectors. Number of outputs = the total number of labels.\n",
    "        features_size = train_train.select(\"features\").first()[0].size\n",
    "        model = classifier(layers=[features_size,10,2]) \n",
    "    model_trained = model.fit(train_train)\n",
    "\n",
    "    train_test_predicted = model_trained.transform(train_test)\n",
    "\n",
    "    evaluator = MulticlassClassificationEvaluator(labelCol=\"label\", predictionCol=\"prediction\") \n",
    "    accuracy = evaluator.evaluate(train_test_predicted, {evaluator.metricName: \"accuracy\"}) # f1|weightedPrecision|weightedRecall|accuracy\n",
    "    print(\"\"+classifier.__name__.ljust(30) + '\\t' + str(accuracy))\n",
    "#     test_predicted.show(10)\n",
    "\n",
    "#     print('Wrong predictions for error analysis')\n",
    "#     test_predicted.filter(test_predicted['prediction'] != test_predicted['label']).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Working on Test data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------+--------------------+------+----+-----+-----+-------+-------+-----+--------+\n",
      "|PassengerId|Pclass|                Name|   Sex| Age|SibSp|Parch| Ticket|   Fare|Cabin|Embarked|\n",
      "+-----------+------+--------------------+------+----+-----+-----+-------+-------+-----+--------+\n",
      "|        892|     3|    Kelly, Mr. James|  male|34.5|    0|    0| 330911| 7.8292| null|       Q|\n",
      "|        893|     3|Wilkes, Mrs. Jame...|female|47.0|    1|    0| 363272|    7.0| null|       S|\n",
      "|        894|     2|Myles, Mr. Thomas...|  male|62.0|    0|    0| 240276| 9.6875| null|       Q|\n",
      "|        895|     3|    Wirz, Mr. Albert|  male|27.0|    0|    0| 315154| 8.6625| null|       S|\n",
      "|        896|     3|Hirvonen, Mrs. Al...|female|22.0|    1|    1|3101298|12.2875| null|       S|\n",
      "+-----------+------+--------------------+------+----+-----+-----+-------+-------+-----+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[PassengerId: int, Pclass: int, Name: string, Sex: int, Age: double, SibSp: int, Parch: int, Ticket: string, Fare: double, Cabin: int, Embarked: int]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_data = (sqlContext\n",
    "                 .read\n",
    "                 .format('csv')\n",
    "                 .options(header='true', inferSchema='true')\n",
    "                 .load('./data/test.csv'))\n",
    "test_data.cache()\n",
    "test_data.show(5)\n",
    "\n",
    "test_data = prepare_data(test_data)\n",
    "test_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------------------------------+\n",
      "|PassengerId|features                         |\n",
      "+-----------+---------------------------------+\n",
      "|892        |[3.0,0.0,34.5,0.0,0.0,7.8292,2.0]|\n",
      "+-----------+---------------------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_assembler = VectorAssembler(\n",
    "    inputCols=['Pclass','Sex','Age','SibSp','Parch','Fare','Embarked'], outputCol='features'\n",
    ")\n",
    "\n",
    "test_data = test_assembler.transform(test_data).select('PassengerId','features')\n",
    "test_data.show(1,truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Proceed with Random Forest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RandomForestClassificationModel (uid=rfc_4aa3cfd71a7a) with 20 trees"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rf_model = RandomForestClassifier()\n",
    "rf_model_fitted = rf_model.fit(train_data)\n",
    "rf_model_fitted"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------------------------------+--------------------------------------+---------------------------------------+----------+\n",
      "|PassengerId|features                         |rawPrediction                         |probability                            |prediction|\n",
      "+-----------+---------------------------------+--------------------------------------+---------------------------------------+----------+\n",
      "|892        |[3.0,0.0,34.5,0.0,0.0,7.8292,2.0]|[18.05041587222414,1.9495841277758608]|[0.902520793611207,0.09747920638879304]|0.0       |\n",
      "+-----------+---------------------------------+--------------------------------------+---------------------------------------+----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rf_test_predicted = rf_model_fitted.transform(test_data)\n",
    "rf_test_predicted.show(1,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rf_test_predicted.select(\"PassengerId\", col(\"prediction\").alias('Survived')).write.csv(\"output\",mode='overwrite',header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cluster most probabable survivors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+--------------------+-----------+----------+\n",
      "|PassengerId|            features|       rawPrediction|probability|prediction|\n",
      "+-----------+--------------------+--------------------+-----------+----------+\n",
      "|        892|[3.0,0.0,34.5,0.0...|[18.0504158722241...| 0.09747921|       0.0|\n",
      "+-----------+--------------------+--------------------+-----------+----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def extract_survival_prob(probability):\n",
    "    return probability.values[1].item() \n",
    "\n",
    "extract_survival_prob_udf = functions.udf(extract_survival_prob, FloatType())\n",
    "\n",
    "survival_prob_udf_df = rf_test_predicted.select('*', extract_survival_prob_udf(rf_test_predicted['probability']).alias('survival_prob_udf'))\n",
    "# survival_prob_udf_df.show(1)\n",
    "\n",
    "survival_prob_df = survival_prob_udf_df.withColumn('probability', survival_prob_udf_df['survival_prob_udf']).drop('survival_prob_udf').cache()\n",
    "survival_prob_df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+\n",
      "|PassengerId|            features|\n",
      "+-----------+--------------------+\n",
      "|        904|[1.0,1.0,23.0,1.0...|\n",
      "|        906|[1.0,1.0,47.0,1.0...|\n",
      "|        907|[2.0,1.0,24.0,1.0...|\n",
      "|        914|[1.0,1.0,40.91836...|\n",
      "|        916|[1.0,1.0,48.0,1.0...|\n",
      "+-----------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "likely_survivors_df = survival_prob_df.filter(\"probability > 0.8\")\n",
    "\n",
    "likely_survivors_df=likely_survivors_df.drop('probability','rawPrediction','prediction')\n",
    "likely_survivors_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['Pclass', 'Sex', 'Age', 'SibSp', 'Parch', 'Fare', 'Embarked']\n",
      "+---------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|mean                                                                                                                                   |\n",
      "+---------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|[2.0882353113706267,0.8823531137782782,21.613472813187634,0.5294116726332617,0.8235293772121892,25.108456293919506,2.7647058363349806] |\n",
      "|[1.0000001386082993,0.9999998614345726,41.204926539856274,0.5000000692945482,0.39583341708660036,109.44912135220906,1.8125001645715564]|\n",
      "+---------------------------------------------------------------------------------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.clustering import GaussianMixture\n",
    "\n",
    "gmm = GaussianMixture().setK(2).setSeed(607262)\n",
    "gmm_model = gmm.fit(likely_survivors_df)\n",
    "print assembler.getInputCols()\n",
    "gmm_model.gaussiansDF.select('mean').show(5, truncate=False)\n",
    "# gmm_model.gaussiansDF.write.csv(\"output\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0.4146340652098827, 0.5853659347901172]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gmm_model.weights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[34, 48]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gmm_model.summary.clusterSizes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [default]",
   "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"
  },
  "name": "Titanic Tutorial",
  "notebookId": 338688711146068
 },
 "nbformat": 4,
 "nbformat_minor": 0
}