titanic_survival_model.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.1.53:4041\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.6</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>PySparkShell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x108f89518>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Loading the data into Spark using DataFrames\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('Titanic Data').getOrCreate()\n",
    "\n",
    "spark # prints what's in the object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[PassengerId: string, Survived: string, Pclass: string, Name: string, Sex: string, Age: string, SibSp: string, Parch: string, Ticket: string, Fare: string, Cabin: string, Embarked: string]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Read the training data and create a DataFrames\n",
    "df = (spark.read\n",
    "          .format(\"csv\")\n",
    "          .option('header', 'true')\n",
    "          .load(\"./data/train.csv\"))\n",
    "\n",
    "df # prints the DF structure"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "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 ...|  male|  22|    1|    0|       A/5 21171|   7.25| null|       S|\n",
      "|          2|       1|     1|Cumings, Mrs. Joh...|female|  38|    1|    0|        PC 17599|71.2833|  C85|       C|\n",
      "|          3|       1|     3|Heikkinen, Miss. ...|female|  26|    0|    0|STON/O2. 3101282|  7.925| null|       S|\n",
      "|          4|       1|     1|Futrelle, Mrs. Ja...|female|  35|    1|    0|          113803|   53.1| C123|       S|\n",
      "|          5|       0|     3|Allen, Mr. Willia...|  male|  35|    0|    0|          373450|   8.05| null|       S|\n",
      "|          6|       0|     3|    Moran, Mr. James|  male|null|    0|    0|          330877| 8.4583| null|       Q|\n",
      "|          7|       0|     1|McCarthy, Mr. Tim...|  male|  54|    0|    0|           17463|51.8625|  E46|       S|\n",
      "|          8|       0|     3|Palsson, Master. ...|  male|   2|    3|    1|          349909| 21.075| null|       S|\n",
      "|          9|       1|     3|Johnson, Mrs. Osc...|female|  27|    0|    2|          347742|11.1333| null|       S|\n",
      "|         10|       1|     2|Nasser, Mrs. Nich...|female|  14|    1|    0|          237736|30.0708| null|       C|\n",
      "|         11|       1|     3|Sandstrom, Miss. ...|female|   4|    1|    1|         PP 9549|   16.7|   G6|       S|\n",
      "|         12|       1|     1|Bonnell, Miss. El...|female|  58|    0|    0|          113783|  26.55| C103|       S|\n",
      "|         13|       0|     3|Saundercock, Mr. ...|  male|  20|    0|    0|       A/5. 2151|   8.05| null|       S|\n",
      "|         14|       0|     3|Andersson, Mr. An...|  male|  39|    1|    5|          347082| 31.275| null|       S|\n",
      "|         15|       0|     3|Vestrom, Miss. Hu...|female|  14|    0|    0|          350406| 7.8542| null|       S|\n",
      "|         16|       1|     2|Hewlett, Mrs. (Ma...|female|  55|    0|    0|          248706|     16| null|       S|\n",
      "|         17|       0|     3|Rice, Master. Eugene|  male|   2|    4|    1|          382652| 29.125| null|       Q|\n",
      "|         18|       1|     2|Williams, Mr. Cha...|  male|null|    0|    0|          244373|     13| null|       S|\n",
      "|         19|       0|     3|Vander Planke, Mr...|female|  31|    1|    0|          345763|     18| null|       S|\n",
      "|         20|       1|     3|Masselmani, Mrs. ...|female|null|    0|    0|            2649|  7.225| null|       C|\n",
      "+-----------+--------+------+--------------------+------+----+-----+-----+----------------+-------+-----+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "891"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.show() # prints some records from the DF\n",
    "df.count() # prints count of records in the training data\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+------+----+-------+--------+\n",
      "|Survived|Pclass|   Sex| Age|   Fare|Embarked|\n",
      "+--------+------+------+----+-------+--------+\n",
      "|     0.0|   3.0|  male|22.0|   7.25|       S|\n",
      "|     1.0|   1.0|female|38.0|71.2833|       C|\n",
      "|     1.0|   3.0|female|26.0|  7.925|       S|\n",
      "|     1.0|   1.0|female|35.0|   53.1|       S|\n",
      "|     0.0|   3.0|  male|35.0|   8.05|       S|\n",
      "|     0.0|   3.0|  male|null| 8.4583|       Q|\n",
      "|     0.0|   1.0|  male|54.0|51.8625|       S|\n",
      "|     0.0|   3.0|  male| 2.0| 21.075|       S|\n",
      "|     1.0|   3.0|female|27.0|11.1333|       S|\n",
      "|     1.0|   2.0|female|14.0|30.0708|       C|\n",
      "|     1.0|   3.0|female| 4.0|   16.7|       S|\n",
      "|     1.0|   1.0|female|58.0|  26.55|       S|\n",
      "|     0.0|   3.0|  male|20.0|   8.05|       S|\n",
      "|     0.0|   3.0|  male|39.0| 31.275|       S|\n",
      "|     0.0|   3.0|female|14.0| 7.8542|       S|\n",
      "|     1.0|   2.0|female|55.0|   16.0|       S|\n",
      "|     0.0|   3.0|  male| 2.0| 29.125|       Q|\n",
      "|     1.0|   2.0|  male|null|   13.0|       S|\n",
      "|     0.0|   3.0|female|31.0|   18.0|       S|\n",
      "|     1.0|   3.0|female|null|  7.225|       C|\n",
      "+--------+------+------+----+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Prepare a dataset by casting some of columns to required type!\n",
    "\n",
    "from pyspark.sql.functions import col\n",
    "dataset = df.select(col('Survived').cast('float'),\n",
    "                         col('Pclass').cast('float'),\n",
    "                         col('Sex'),\n",
    "                         col('Age').cast('float'),\n",
    "                         col('Fare').cast('float'),\n",
    "                         col('Embarked')\n",
    "                        )\n",
    "dataset.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+---+---+----+--------+\n",
      "|Survived|Pclass|Sex|Age|Fare|Embarked|\n",
      "+--------+------+---+---+----+--------+\n",
      "|       0|     0|  0|177|   0|       2|\n",
      "+--------+------+---+---+----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import isnull, when, count, col\n",
    "dataset.select([count(when(isnull(c), c)).alias(c) for c in dataset.columns]).show() # todo: learn to query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+------+----+-------+--------+\n",
      "|Survived|Pclass|   Sex| Age|   Fare|Embarked|\n",
      "+--------+------+------+----+-------+--------+\n",
      "|     0.0|   3.0|  male|22.0|   7.25|       S|\n",
      "|     1.0|   1.0|female|38.0|71.2833|       C|\n",
      "|     1.0|   3.0|female|26.0|  7.925|       S|\n",
      "|     1.0|   1.0|female|35.0|   53.1|       S|\n",
      "|     0.0|   3.0|  male|35.0|   8.05|       S|\n",
      "|     0.0|   1.0|  male|54.0|51.8625|       S|\n",
      "|     0.0|   3.0|  male| 2.0| 21.075|       S|\n",
      "|     1.0|   3.0|female|27.0|11.1333|       S|\n",
      "|     1.0|   2.0|female|14.0|30.0708|       C|\n",
      "|     1.0|   3.0|female| 4.0|   16.7|       S|\n",
      "|     1.0|   1.0|female|58.0|  26.55|       S|\n",
      "|     0.0|   3.0|  male|20.0|   8.05|       S|\n",
      "|     0.0|   3.0|  male|39.0| 31.275|       S|\n",
      "|     0.0|   3.0|female|14.0| 7.8542|       S|\n",
      "|     1.0|   2.0|female|55.0|   16.0|       S|\n",
      "|     0.0|   3.0|  male| 2.0| 29.125|       Q|\n",
      "|     0.0|   3.0|female|31.0|   18.0|       S|\n",
      "|     0.0|   2.0|  male|35.0|   26.0|       S|\n",
      "|     1.0|   2.0|  male|34.0|   13.0|       S|\n",
      "|     1.0|   3.0|female|15.0| 8.0292|       Q|\n",
      "+--------+------+------+----+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "712"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Eliminate rows with null values in columns\n",
    "dataset = dataset.replace('?', None).dropna(how='any') # todo: learn the Dataset API\n",
    "dataset.show()\n",
    "dataset.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+------+----+-------+--------+------+-------+\n",
      "|Survived|Pclass|   Sex| Age|   Fare|Embarked|Gender|Boarded|\n",
      "+--------+------+------+----+-------+--------+------+-------+\n",
      "|     0.0|   3.0|  male|22.0|   7.25|       S|   0.0|    0.0|\n",
      "|     1.0|   1.0|female|38.0|71.2833|       C|   1.0|    1.0|\n",
      "|     1.0|   3.0|female|26.0|  7.925|       S|   1.0|    0.0|\n",
      "|     1.0|   1.0|female|35.0|   53.1|       S|   1.0|    0.0|\n",
      "|     0.0|   3.0|  male|35.0|   8.05|       S|   0.0|    0.0|\n",
      "|     0.0|   1.0|  male|54.0|51.8625|       S|   0.0|    0.0|\n",
      "|     0.0|   3.0|  male| 2.0| 21.075|       S|   0.0|    0.0|\n",
      "|     1.0|   3.0|female|27.0|11.1333|       S|   1.0|    0.0|\n",
      "|     1.0|   2.0|female|14.0|30.0708|       C|   1.0|    1.0|\n",
      "|     1.0|   3.0|female| 4.0|   16.7|       S|   1.0|    0.0|\n",
      "|     1.0|   1.0|female|58.0|  26.55|       S|   1.0|    0.0|\n",
      "|     0.0|   3.0|  male|20.0|   8.05|       S|   0.0|    0.0|\n",
      "|     0.0|   3.0|  male|39.0| 31.275|       S|   0.0|    0.0|\n",
      "|     0.0|   3.0|female|14.0| 7.8542|       S|   1.0|    0.0|\n",
      "|     1.0|   2.0|female|55.0|   16.0|       S|   1.0|    0.0|\n",
      "|     0.0|   3.0|  male| 2.0| 29.125|       Q|   0.0|    2.0|\n",
      "|     0.0|   3.0|female|31.0|   18.0|       S|   1.0|    0.0|\n",
      "|     0.0|   2.0|  male|35.0|   26.0|       S|   0.0|    0.0|\n",
      "|     1.0|   2.0|  male|34.0|   13.0|       S|   0.0|    0.0|\n",
      "|     1.0|   3.0|female|15.0| 8.0292|       Q|   1.0|    2.0|\n",
      "+--------+------+------+----+-------+--------+------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Spark ML library only works with numeric data. \n",
    "# But we still want to use the Sex and the Embarked column. \n",
    "# For that, we will need to encode (transform) them. Sex -> Gender; Embarked -> Boarded\n",
    "\n",
    "from pyspark.ml.feature import StringIndexer # todo: learn Spark ML API\n",
    "dataset = StringIndexer(\n",
    "    inputCol='Sex', \n",
    "    outputCol='Gender', \n",
    "    handleInvalid='keep').fit(dataset).transform(dataset)\n",
    "dataset = StringIndexer(\n",
    "    inputCol='Embarked', \n",
    "    outputCol='Boarded', \n",
    "    handleInvalid='keep').fit(dataset).transform(dataset)\n",
    "\n",
    "dataset.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Survived', 'float'),\n",
       " ('Pclass', 'float'),\n",
       " ('Sex', 'string'),\n",
       " ('Age', 'float'),\n",
       " ('Fare', 'float'),\n",
       " ('Embarked', 'string'),\n",
       " ('Gender', 'double'),\n",
       " ('Boarded', 'double')]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dataset.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+----+-------+------+-------+\n",
      "|Survived|Pclass| Age|   Fare|Gender|Boarded|\n",
      "+--------+------+----+-------+------+-------+\n",
      "|     0.0|   3.0|22.0|   7.25|   0.0|    0.0|\n",
      "|     1.0|   1.0|38.0|71.2833|   1.0|    1.0|\n",
      "|     1.0|   3.0|26.0|  7.925|   1.0|    0.0|\n",
      "|     1.0|   1.0|35.0|   53.1|   1.0|    0.0|\n",
      "|     0.0|   3.0|35.0|   8.05|   0.0|    0.0|\n",
      "|     0.0|   1.0|54.0|51.8625|   0.0|    0.0|\n",
      "|     0.0|   3.0| 2.0| 21.075|   0.0|    0.0|\n",
      "|     1.0|   3.0|27.0|11.1333|   1.0|    0.0|\n",
      "|     1.0|   2.0|14.0|30.0708|   1.0|    1.0|\n",
      "|     1.0|   3.0| 4.0|   16.7|   1.0|    0.0|\n",
      "|     1.0|   1.0|58.0|  26.55|   1.0|    0.0|\n",
      "|     0.0|   3.0|20.0|   8.05|   0.0|    0.0|\n",
      "|     0.0|   3.0|39.0| 31.275|   0.0|    0.0|\n",
      "|     0.0|   3.0|14.0| 7.8542|   1.0|    0.0|\n",
      "|     1.0|   2.0|55.0|   16.0|   1.0|    0.0|\n",
      "|     0.0|   3.0| 2.0| 29.125|   0.0|    2.0|\n",
      "|     0.0|   3.0|31.0|   18.0|   1.0|    0.0|\n",
      "|     0.0|   2.0|35.0|   26.0|   0.0|    0.0|\n",
      "|     1.0|   2.0|34.0|   13.0|   0.0|    0.0|\n",
      "|     1.0|   3.0|15.0| 8.0292|   1.0|    2.0|\n",
      "+--------+------+----+-------+------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "712"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Drop unnecessary columns\n",
    "dataset = dataset.drop('Sex')\n",
    "dataset = dataset.drop('Embarked')\n",
    "dataset.show()\n",
    "dataset.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Survived: float, Pclass: float, Age: float, Fare: float, Gender: double, Boarded: double, features: vector]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Spark works to predict with a column with all the features smashed together into a list-like structure.\n",
    "# I want to predict “Survived”, I need to combine the information other columns into one column.\n",
    "# That column is called \"features\" and it's value should look like say [3.0, 22.0, 7.25, 0, 0]\n",
    "\n",
    "# Assemble all the features with VectorAssembler\n",
    "required_features = ['Pclass',\n",
    "                    'Age',\n",
    "                    'Fare',\n",
    "                    'Gender',\n",
    "                    'Boarded'\n",
    "                   ]\n",
    "from pyspark.ml.feature import VectorAssembler # todo\n",
    "assembler = VectorAssembler(inputCols=required_features, outputCol='features')\n",
    "transformed_data = assembler.transform(dataset)\n",
    "\n",
    "transformed_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "|Survived|Pclass| Age|   Fare|Gender|Boarded|            features|\n",
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "|     0.0|   3.0|22.0|   7.25|   0.0|    0.0|[3.0,22.0,7.25,0....|\n",
      "|     1.0|   1.0|38.0|71.2833|   1.0|    1.0|[1.0,38.0,71.2833...|\n",
      "|     1.0|   3.0|26.0|  7.925|   1.0|    0.0|[3.0,26.0,7.92500...|\n",
      "|     1.0|   1.0|35.0|   53.1|   1.0|    0.0|[1.0,35.0,53.0999...|\n",
      "|     0.0|   3.0|35.0|   8.05|   0.0|    0.0|[3.0,35.0,8.05000...|\n",
      "|     0.0|   1.0|54.0|51.8625|   0.0|    0.0|[1.0,54.0,51.8624...|\n",
      "|     0.0|   3.0| 2.0| 21.075|   0.0|    0.0|[3.0,2.0,21.07500...|\n",
      "|     1.0|   3.0|27.0|11.1333|   1.0|    0.0|[3.0,27.0,11.1332...|\n",
      "|     1.0|   2.0|14.0|30.0708|   1.0|    1.0|[2.0,14.0,30.0708...|\n",
      "|     1.0|   3.0| 4.0|   16.7|   1.0|    0.0|[3.0,4.0,16.70000...|\n",
      "|     1.0|   1.0|58.0|  26.55|   1.0|    0.0|[1.0,58.0,26.5499...|\n",
      "|     0.0|   3.0|20.0|   8.05|   0.0|    0.0|[3.0,20.0,8.05000...|\n",
      "|     0.0|   3.0|39.0| 31.275|   0.0|    0.0|[3.0,39.0,31.2749...|\n",
      "|     0.0|   3.0|14.0| 7.8542|   1.0|    0.0|[3.0,14.0,7.85419...|\n",
      "|     1.0|   2.0|55.0|   16.0|   1.0|    0.0|[2.0,55.0,16.0,1....|\n",
      "|     0.0|   3.0| 2.0| 29.125|   0.0|    2.0|[3.0,2.0,29.125,0...|\n",
      "|     0.0|   3.0|31.0|   18.0|   1.0|    0.0|[3.0,31.0,18.0,1....|\n",
      "|     0.0|   2.0|35.0|   26.0|   0.0|    0.0|[2.0,35.0,26.0,0....|\n",
      "|     1.0|   2.0|34.0|   13.0|   0.0|    0.0|[2.0,34.0,13.0,0....|\n",
      "|     1.0|   3.0|15.0| 8.0292|   1.0|    2.0|[3.0,15.0,8.02919...|\n",
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "712"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "transformed_data.show()\n",
    "transformed_data.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# At this point our data prep is done\n",
    "# We will start Modeling now.."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Before modeling let’s do the usual splitting between training and testing\n",
    "(training_data, test_data) = transformed_data.randomSplit([0.8,0.2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+----+--------+------+-------+--------------------+\n",
      "|Survived|Pclass| Age|    Fare|Gender|Boarded|            features|\n",
      "+--------+------+----+--------+------+-------+--------------------+\n",
      "|     0.0|   1.0|19.0|    53.1|   0.0|    0.0|[1.0,19.0,53.0999...|\n",
      "|     0.0|   1.0|19.0|   263.0|   0.0|    0.0|[1.0,19.0,263.0,0...|\n",
      "|     0.0|   1.0|21.0| 77.2875|   0.0|    0.0|[1.0,21.0,77.2874...|\n",
      "|     0.0|   1.0|22.0|135.6333|   0.0|    1.0|[1.0,22.0,135.633...|\n",
      "|     0.0|   1.0|24.0|247.5208|   0.0|    1.0|[1.0,24.0,247.520...|\n",
      "|     0.0|   1.0|25.0|  151.55|   1.0|    0.0|[1.0,25.0,151.550...|\n",
      "|     0.0|   1.0|27.0|   211.5|   0.0|    1.0|[1.0,27.0,211.5,0...|\n",
      "|     0.0|   1.0|28.0| 82.1708|   0.0|    1.0|[1.0,28.0,82.1707...|\n",
      "|     0.0|   1.0|29.0|    30.0|   0.0|    0.0|[1.0,29.0,30.0,0....|\n",
      "|     0.0|   1.0|29.0|    66.6|   0.0|    0.0|[1.0,29.0,66.5999...|\n",
      "|     0.0|   1.0|30.0|   27.75|   0.0|    1.0|[1.0,30.0,27.75,0...|\n",
      "|     0.0|   1.0|31.0| 50.4958|   0.0|    0.0|[1.0,31.0,50.4958...|\n",
      "|     0.0|   1.0|31.0|    52.0|   0.0|    0.0|[1.0,31.0,52.0,0....|\n",
      "|     0.0|   1.0|33.0|     5.0|   0.0|    0.0|[1.0,33.0,5.0,0.0...|\n",
      "|     0.0|   1.0|36.0|  40.125|   0.0|    1.0|[1.0,36.0,40.125,...|\n",
      "|     0.0|   1.0|36.0|   78.85|   0.0|    0.0|[1.0,36.0,78.8499...|\n",
      "|     0.0|   1.0|37.0|    53.1|   0.0|    0.0|[1.0,37.0,53.0999...|\n",
      "|     0.0|   1.0|38.0|     0.0|   0.0|    0.0|(5,[0,1],[1.0,38.0])|\n",
      "|     0.0|   1.0|38.0|153.4625|   0.0|    0.0|[1.0,38.0,153.462...|\n",
      "|     0.0|   1.0|39.0|     0.0|   0.0|    0.0|(5,[0,1],[1.0,39.0])|\n",
      "+--------+------+----+--------+------+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "562"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# We randomly select 80% of the transformed data and use it as training_data\n",
    "training_data.show()\n",
    "training_data.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "|Survived|Pclass| Age|   Fare|Gender|Boarded|            features|\n",
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "|     0.0|   1.0| 2.0| 151.55|   1.0|    0.0|[1.0,2.0,151.5500...|\n",
      "|     0.0|   1.0|18.0|  108.9|   0.0|    1.0|[1.0,18.0,108.900...|\n",
      "|     0.0|   1.0|24.0|   79.2|   0.0|    1.0|[1.0,24.0,79.1999...|\n",
      "|     0.0|   1.0|28.0|   47.1|   0.0|    0.0|[1.0,28.0,47.0999...|\n",
      "|     0.0|   1.0|37.0|   29.7|   0.0|    1.0|[1.0,37.0,29.7000...|\n",
      "|     0.0|   1.0|45.0|   35.5|   0.0|    0.0|[1.0,45.0,35.5,0....|\n",
      "|     0.0|   1.0|46.0|   79.2|   0.0|    1.0|[1.0,46.0,79.1999...|\n",
      "|     0.0|   1.0|47.0|25.5875|   0.0|    0.0|[1.0,47.0,25.5874...|\n",
      "|     0.0|   1.0|50.0|106.425|   0.0|    1.0|[1.0,50.0,106.425...|\n",
      "|     0.0|   1.0|54.0|51.8625|   0.0|    0.0|[1.0,54.0,51.8624...|\n",
      "|     0.0|   1.0|62.0|  26.55|   0.0|    0.0|[1.0,62.0,26.5499...|\n",
      "|     0.0|   1.0|64.0|   26.0|   0.0|    0.0|[1.0,64.0,26.0,0....|\n",
      "|     0.0|   1.0|71.0|49.5042|   0.0|    1.0|[1.0,71.0,49.5041...|\n",
      "|     0.0|   2.0|21.0|   11.5|   0.0|    0.0|[2.0,21.0,11.5,0....|\n",
      "|     0.0|   2.0|23.0|   13.0|   0.0|    0.0|[2.0,23.0,13.0,0....|\n",
      "|     0.0|   2.0|23.0|   13.0|   0.0|    0.0|[2.0,23.0,13.0,0....|\n",
      "|     0.0|   2.0|24.0|   13.0|   1.0|    0.0|[2.0,24.0,13.0,1....|\n",
      "|     0.0|   2.0|25.0|   13.0|   0.0|    0.0|[2.0,25.0,13.0,0....|\n",
      "|     0.0|   2.0|25.0|   26.0|   0.0|    0.0|[2.0,25.0,26.0,0....|\n",
      "|     0.0|   2.0|26.0|   26.0|   1.0|    0.0|[2.0,26.0,26.0,1....|\n",
      "+--------+------+----+-------+------+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "150"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# remaining 20% as test_data\n",
    "\n",
    "test_data.show()\n",
    "test_data.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Build and fit an ML model to our dataset to predict the “Survived” columns with all the other ones. \n",
    "# We will be using a Random Forest Classifier. This is actually an estimator that we have to fit.\n",
    "\n",
    "from pyspark.ml.classification import RandomForestClassifier # todo\n",
    "rf = RandomForestClassifier(labelCol='Survived', \n",
    "                            featuresCol='features',\n",
    "                            maxDepth=5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RandomForestClassificationModel (uid=RandomForestClassifier_9c4972e3b993) with 20 trees"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Now we fit the model\n",
    "model = rf.fit(training_data)\n",
    "\n",
    "# This will give us something called a transformer.\n",
    "\n",
    "model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Survived: float, Pclass: float, Age: float, Fare: float, Gender: double, Boarded: double, features: vector, rawPrediction: vector, probability: vector, prediction: double]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# And finally, we predict using the test dataset\n",
    "predictions = model.transform(test_data)\n",
    "\n",
    "predictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+----+-------+------+-------+--------------------+--------------------+--------------------+----------+\n",
      "|Survived|Pclass| Age|   Fare|Gender|Boarded|            features|       rawPrediction|         probability|prediction|\n",
      "+--------+------+----+-------+------+-------+--------------------+--------------------+--------------------+----------+\n",
      "|     0.0|   1.0| 2.0| 151.55|   1.0|    0.0|[1.0,2.0,151.5500...|[1.03384693241575...|[0.05169234662078...|       1.0|\n",
      "|     0.0|   1.0|18.0|  108.9|   0.0|    1.0|[1.0,18.0,108.900...|[11.3766016777891...|[0.56883008388945...|       0.0|\n",
      "|     0.0|   1.0|24.0|   79.2|   0.0|    1.0|[1.0,24.0,79.1999...|[10.2970030600326...|[0.51485015300163...|       0.0|\n",
      "|     0.0|   1.0|28.0|   47.1|   0.0|    0.0|[1.0,28.0,47.0999...|[15.4988687426680...|[0.77494343713340...|       0.0|\n",
      "|     0.0|   1.0|37.0|   29.7|   0.0|    1.0|[1.0,37.0,29.7000...|[14.9841949332446...|[0.74920974666223...|       0.0|\n",
      "|     0.0|   1.0|45.0|   35.5|   0.0|    0.0|[1.0,45.0,35.5,0....|[15.3335500845210...|[0.76667750422605...|       0.0|\n",
      "|     0.0|   1.0|46.0|   79.2|   0.0|    1.0|[1.0,46.0,79.1999...|[11.4598835129131...|[0.57299417564565...|       0.0|\n",
      "|     0.0|   1.0|47.0|25.5875|   0.0|    0.0|[1.0,47.0,25.5874...|[16.5978310034312...|[0.82989155017156...|       0.0|\n",
      "|     0.0|   1.0|50.0|106.425|   0.0|    1.0|[1.0,50.0,106.425...|[12.7111993023868...|[0.63555996511934...|       0.0|\n",
      "|     0.0|   1.0|54.0|51.8625|   0.0|    0.0|[1.0,54.0,51.8624...|[15.2820145029054...|[0.76410072514527...|       0.0|\n",
      "|     0.0|   1.0|62.0|  26.55|   0.0|    0.0|[1.0,62.0,26.5499...|[16.7203606477000...|[0.83601803238500...|       0.0|\n",
      "|     0.0|   1.0|64.0|   26.0|   0.0|    0.0|[1.0,64.0,26.0,0....|[16.5978310034312...|[0.82989155017156...|       0.0|\n",
      "|     0.0|   1.0|71.0|49.5042|   0.0|    1.0|[1.0,71.0,49.5041...|[15.3067670755024...|[0.76533835377512...|       0.0|\n",
      "|     0.0|   2.0|21.0|   11.5|   0.0|    0.0|[2.0,21.0,11.5,0....|[17.6782667603222...|[0.88391333801610...|       0.0|\n",
      "|     0.0|   2.0|23.0|   13.0|   0.0|    0.0|[2.0,23.0,13.0,0....|[17.5998539382397...|[0.87999269691198...|       0.0|\n",
      "|     0.0|   2.0|23.0|   13.0|   0.0|    0.0|[2.0,23.0,13.0,0....|[17.5998539382397...|[0.87999269691198...|       0.0|\n",
      "|     0.0|   2.0|24.0|   13.0|   1.0|    0.0|[2.0,24.0,13.0,1....|[1.00013910814984...|[0.05000695540749...|       1.0|\n",
      "|     0.0|   2.0|25.0|   13.0|   0.0|    0.0|[2.0,25.0,13.0,0....|[17.5998539382397...|[0.87999269691198...|       0.0|\n",
      "|     0.0|   2.0|25.0|   26.0|   0.0|    0.0|[2.0,25.0,26.0,0....|[17.5319034013526...|[0.87659517006763...|       0.0|\n",
      "|     0.0|   2.0|26.0|   26.0|   1.0|    0.0|[2.0,26.0,26.0,1....|[1.00442441647607...|[0.05022122082380...|       1.0|\n",
      "+--------+------+----+-------+------+-------+--------------------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Done! My first Spark ML model\n",
    "predictions.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MulticlassClassificationEvaluator_38f388550d7a"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Evaluate the model using a basic metric called the accuracy\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator # todo\n",
    "evaluator = MulticlassClassificationEvaluator(\n",
    "    labelCol='Survived', \n",
    "    predictionCol='prediction', \n",
    "    metricName='accuracy')\n",
    "\n",
    "evaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test Accuracy =  0.7866666666666666\n"
     ]
    }
   ],
   "source": [
    "# And this gives me the accuracy\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print('Test Accuracy = ', accuracy)"
   ]
  }
 ],
 "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
}