jupyter/notebook/titanic/titanic_spark_exercises.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Clasificación con data de hundimiento del Titanic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('pyspark-titanic-exe').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft = spark.read.csv('/dataset/titanic.csv', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import DoubleType, IntegerType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft = dft.withColumn('survived', dft['survived'].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft.select('survived').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dft.groupBy('survived').count().orderBy('count').show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extraccion (Ejercicio 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pathlib import Path\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_titanic_data(url, refresh_cache=False):\n",
    "    \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "url = 'http://biostat.mc.vanderbilt.edu/wiki/pub/Main/DataSets/titanic3.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw = extract_titanic_data(url)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(df_raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Rename home.dest column\n",
    "\n",
    "df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Train/test split\n",
    "train_df, test_df = df_raw.randomSplit([0.7,0.3],seed=1234)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#df_raw.randomSplit?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import rand, when\n",
    "df_raw = df_raw.withColumn('train', when(rand(seed=1234) >= 0.3, True).otherwise(False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#rand?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.select('train').groupby('train').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.select('train').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train = df_raw.filter(f.col('train') == True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test = df_raw.filter(f.col('train') != True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 2: Primer preproceso / EDA"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Casteo de datos\n",
    "df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cast_cols(df, cols, new_type):\n",
    " \n",
    "  return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "integer_cols = ['survived', 'sibsp', 'parch', 'body']\n",
    "float_cols = ['age', 'fare']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw = cast_cols(df_raw, integer_cols, IntegerType)\n",
    "df_raw = cast_cols(df_raw, float_cols, DoubleType)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_raw.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "survived_on_boat = df_raw.filter((f.col('boat').isNotNull()) & (f.col('survived') == 1)).count()\n",
    "survived = df_raw.filter(f.col('survived') == 1).count()\n",
    "survived_on_boat / survived"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "logging.basicConfig(\n",
    "    level=logging.INFO,\n",
    "    format='%(asctime)s-%(name)s-%(levelname)s: %(message)s',\n",
    "    handlers=[logging.FileHandler('/dataset/titanic_spark.log'), logging.StreamHandler()],\n",
    ")\n",
    "logger = logging.getLogger(__name__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _drop_unusable_cols():\n",
    "    logger.info(\n",
    "        f\"Dropping the following {len(cols)} unusable columns:\\n\"\n",
    "        f\"{cols}\"\n",
    "    )\n",
    "    \n",
    "    logger.info(\n",
    "        f\"Remaining {len(df.columns)} columns:\\n {sorted(df.columns)}\"\n",
    "    )\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df = _drop_unusable_cols(train_df, cols=['boat', 'body', 'train'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df.columns, test_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_df = _drop_unusable_cols(test_df, cols=['boat', 'body', 'train'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df = train_df.withColumn('train_new', f.lit(True))\n",
    "test_df = test_df.withColumn('train_new', f.lit(False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined_df = train_df.unionByName(test_df)\n",
    "joined_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = _drop_unusable_cols(df_raw, cols=['boat', 'body', 'ticket'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Plotting\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('survived').groupBy('survived').count().toPandas().set_index('survived').plot.bar()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## EDA y Ejercicio 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import seaborn as sns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.heatmap(df.select('survived', 'age', 'parch', 'fare', 'sibsp').toPandas().corr(),\n",
    "                annot=True, fmt = \".2f\", cmap = \"coolwarm\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.FacetGrid(df.toPandas(), col='survived')\n",
    "g = g.map(sns.distplot, 'age')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "survived_age = df.filter((f.col('survived') == 1) & \n",
    "                         (f.col('age').isNotNull())).select('age').toPandas()\n",
    "not_survived_age = df.filter((f.col('survived') == 0) & (f.col('age').isNotNull())).select('age').toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "survived_age.squeeze()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(survived_age.squeeze())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.kdeplot(not_survived_age.squeeze(), color='Red', shade=True)\n",
    "g = sns.kdeplot(survived_age.squeeze(), color='Blue', shade=True)\n",
    "g.set_xlabel('age')\n",
    "g.set_ylabel('Frequency')\n",
    "g.legend(['Not Survived', 'Survived'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fare_mean = df.select(f.mean(f.col('fare'))).first()[0]\n",
    "fare_mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('fare').fillna(fare_mean).filter(f.col('fare').isNull()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Distribución de precio de boletos \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('fare').filter(f.col('fare') == 0).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@f.udf('double')\n",
    "def np_log(x):\n",
    "    if x is None:\n",
    "        return 0\n",
    "    return float(np.log(x + 1)) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('log_fare', np_log(df.fare))\n",
    "df.select('fare', 'log_fare').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop('fare')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.distplot(df.select('log_fare').fillna(np.log(fare_mean)).toPandas(), color='m')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.barplot(x='sex', y='survived', data=df.toPandas())\n",
    "g = g.set_ylabel(\"Survival Probability\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Probabilidad de supervivencia"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.catplot(x='pclass', y='survived', hue='sex', data=df.toPandas(),\n",
    "                   height=6, kind='bar')\n",
    "g = g.set_ylabels(\"survival probability\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 4: Valores nulos y constantes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('age').groupBy('age').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(f.sum(f.col('age').isNull().cast('integer'))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_nulls = df.select([f.sum(f.col(c).isNull().cast('integer') / df.count()).alias(c) for c in df.columns])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_nulls.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "null_cols = [c for c in df_nulls.columns if df_nulls.select(c).first()[0] >= 0.5]\n",
    "null_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _drop_nulls(df, max_null_prop=0.5):\n",
    "    logger.info(\n",
    "        f\"Dropping columns with null ratio greater than {max_null_prop * 100}%...\"\n",
    "    )\n",
    "    df_nulls = df.select([f.sum(f.col(c).isNull().cast('integer') / df.count()).alias(c) \n",
    "                          for c in df.columns])\n",
    "    logger.info(f\"Null proportions:\\n {df_nulls.show()}\")\n",
    "    null_cols = [c for c in df_nulls.columns if df_nulls.select(c).first()[0] > max_null_prop \n",
    "                 and c not in PROTECTED_COLS]\n",
    "    logger.info(f\"Dropping the following {len(null_cols)} columns:\\n {null_cols}\")\n",
    "    df = df.drop(*null_cols)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "PROTECTED_COLS = ['survived', 'train']\n",
    "df = _drop_nulls(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas().isnull().mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas().std()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_cols = [c for c,dtype in df.dtypes if dtype.startswith(('int', 'double'))]\n",
    "num_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_std = df.select([f.stddev(f.col(c)).alias(c) for c in num_cols])\n",
    "df_std.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _drop_std(df, min_std_dev=1.5e-2):\n",
    "   \n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = _drop_std(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _get_typed_cols(df, col_type='cat'):\n",
    "    assert col_type in ('cat', 'num')\n",
    "    dtypes = ('int', 'double') if col_type == 'num' else ('string')\n",
    "    typed_cols = [c for c,dtype in df.dtypes if dtype.startswith(dtypes) \n",
    "                  and c not in PROTECTED_COLS]\n",
    "    return typed_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# _get_typed_cols(df, col_type='foo')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_cols = _get_typed_cols(df, col_type='num')\n",
    "cat_cols = _get_typed_cols(df, col_type='cat')\n",
    "num_cols, cat_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val_counts = df.filter(f.col('embarked').isNotNull()).select('embarked').groupBy('embarked').count().orderBy(f.desc('count'))\n",
    "val_counts.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val_counts.select('embarked').first()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas()['age'].median()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# df.approxQuantile?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.approxQuantile('age', [0.5], 0)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _fill_nulls(df):\n",
    "    for t in ['num', 'cat']:\n",
    "        cols = _get_typed_cols(df, col_type=t)\n",
    "        for c in cols:\n",
    "            if t == 'num':\n",
    "                median_val = df.approxQuantile(c, [0.5], 0)[0]\n",
    "                df = df.fillna(median_val, subset=[c])\n",
    "            else:\n",
    "                val_counts = df.filter(f.col(c).isNotNull()).select(c).groupBy(c).count().orderBy(f.desc('count'))\n",
    "                common_val = val_counts.select(c).first()[0]\n",
    "                df = df.fillna(common_val, subset=[c])\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('embarked').groupBy('embarked').count().orderBy(f.desc('count')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = _fill_nulls(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.toPandas().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('embarked').groupBy('embarked').count().orderBy(f.desc('count')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('age').groupBy('age').count().orderBy(f.desc('count')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 5: Ingenieria de Atributos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('name').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(f.trim(f.split(f.split(f.col('name'), ', ')[1], '. ')[0]).alias('title')).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('title', f.trim(f.split(f.split(df['name'], ', ')[1], '. ')[0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('title').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "title_valcounts = df.select('title').groupBy('title').count().orderBy(f.desc('count'))\n",
    "title_valcounts.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "other_titles = [str(i.title) for i in title_valcounts.select('title').collect()][4:]\n",
    "other_titles"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('title', when(df['title'].isin(*other_titles), 'other').otherwise(df['title']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('title').groupBy('title').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('title', when(df['title'] == 'Miss', 'Mrs').otherwise(df['title']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('title').groupBy('title').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop('name')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.countplot(df.select('title').toPandas().squeeze())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.catplot(x='title',y='survived',data=df.toPandas(),kind=\"bar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Tamaño de familia\n",
    "df.select('parch', 'sibsp', 'family_size').show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "df.select('family_size', 'family_single', 'family_small', 'family_large').show(10)\n",
    "df = df.drop('family_size')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for fsize in ['single', 'small', 'large']:\n",
    "    g = sns.catplot(x=f'family_{fsize}',y='survived',data=df.toPandas(),kind=\"bar\")\n",
    "    g = g.set_ylabels(\"Survival Probability\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 6: Fitteo de regresión logistica"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Data sintetica\n",
    "df1 = spark.createDataFrame([\n",
    "    (1, 10.1, 'a', 'i'),\n",
    "    (0, 14.3, 'b', 'x'),\n",
    "    (0, 3.0, 'c', 'iv'),\n",
    "    (1, 2.5, 'c', 'iv'),\n",
    "    (1, 5.4, 'b', 'i'),\n",
    "    (0, 9.7, 'a', 'x')\n",
    "], ['target', 'numerical', 'cat1', 'cat2'])\n",
    "df1.show()\n",
    "\n",
    "# Convertir strings en numericos\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "\n",
    "cat_cols = ['cat1', 'cat2']\n",
    "for cat in cat_cols:\n",
    "    cat_suff = f'{cat}_num'\n",
    "    if cat_suff not in df1.columns:\n",
    "        indexer = StringIndexer(inputCol=cat, outputCol=cat_suff).fit(df1)\n",
    "        df1 = indexer.transform(df1)\n",
    "df1.show()\n",
    "\n",
    "# Hacer OneHotEnconding\n",
    "from pyspark.ml.feature import OneHotEncoderEstimator\n",
    "\n",
    "encoder = OneHotEncoderEstimator(inputCols=['cat1_num', 'cat2_num'], outputCols=['cat1_vec', 'cat2_vec'])\n",
    "ohem = encoder.fit(df1)\n",
    "df1 = ohem.transform(df1)\n",
    "df1.show()\n",
    "\n",
    "# Armar columna de features\n",
    "df1 = df1.select('target','numerical', 'cat1_vec', 'cat2_vec')\n",
    "df1.show()\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "assembler = VectorAssembler(inputCols=[c for c in df1.columns if c != 'target'], outputCol='features')\n",
    "assembler.transform(df1).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_cols = _get_typed_cols(df, col_type='num')\n",
    "cat_cols = _get_typed_cols(df, col_type='cat')\n",
    "df.columns, cat_cols, num_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StringIndexer\n",
    "\n",
    "def _encode_categorical(df):\n",
    "    cat_cols = _get_typed_cols(df, col_type='cat')\n",
    "    logger.info(f\"Categorical columns:\\n {cat_cols}\")\n",
    "    encoded_cols = []\n",
    "    for cat in cat_cols:\n",
    "        cat_suff = f'{cat}_num'\n",
    "        encoded_cols.append(cat_suff)\n",
    "        if cat_suff not in df.columns:\n",
    "            indexer = StringIndexer(inputCol=cat, outputCol=cat_suff).fit(df)\n",
    "            df = indexer.transform(df)\n",
    "    return df, encoded_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df, encoded_cols = _encode_categorical(df)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_cols = num_cols + encoded_cols\n",
    "feature_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import OneHotEncoderEstimator\n",
    "\n",
    "ohe_cols = [f'{c}_vec' for c in encoded_cols]\n",
    "encoder = OneHotEncoderEstimator(inputCols=encoded_cols, outputCols=ohe_cols)\n",
    "ohem = encoder.fit(df)\n",
    "df = ohem.transform(df)\n",
    "df.show()\n",
    "feature_cols = num_cols + ohe_cols"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "assembler = VectorAssembler(inputCols=feature_cols, outputCol='features')\n",
    "df = assembler.transform(df)\n",
    "df.select('features').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('train').take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data = df.filter(f.col('train') == True).select('survived', 'features')\n",
    "test_data = df.filter(f.col('train') == False).select('survived', 'features')\n",
    "train_data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "lr = LogisticRegression(labelCol='survived', featuresCol='features')\n",
    "lrm = lr.fit(train_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Metricas de evaluacion (insample)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pred_df = lrm.transform(test_data)\n",
    "pred_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "\n",
    "evaluator = BinaryClassificationEvaluator(labelCol='survived')\n",
    "evaluator.evaluate(pred_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 7: árboles"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "pred_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator.evaluate(pred_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "\n",
    "\n",
    "pred_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator.evaluate(pred_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dir(rfm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rfm.featureImportances"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from itertools import chain\n",
    "attrs = sorted(\n",
    "    (attr['idx'], attr['name'])\n",
    "    for attr in (\n",
    "        chain(*pred_df.schema['features'].metadata['ml_attr']['attrs'].values())\n",
    "    )\n",
    ")\n",
    "feat_import = [(name, rfm.featureImportances[idx]) for idx, name in attrs if rfm.featureImportances[idx]]\n",
    "feat_import = pd.DataFrame(feat_import, columns=['feature', 'importance']).sort_values(by='importance', ascending=False)\n",
    "feat_import.head(15)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ax = feat_import[:20].plot(kind='bar')\n",
    "ax.set_xticklabels(feat_import[:20]['feature'].tolist())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}