end_to_end_spark_ml_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark data cleaning and engineering\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'4.9.0'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import plotly\n",
    "plotly.__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[customerID: string, gender: string, SeniorCitizen: string, Partner: string, Dependents: string, tenure: string, PhoneService: string, MultipleLines: string, InternetService: string, OnlineSecurity: string, OnlineBackup: string, DeviceProtection: string, TechSupport: string, StreamingTV: string, StreamingMovies: string, Contract: string, PaperlessBilling: string, PaymentMethod: string, MonthlyCharges: string, TotalCharges: string, Churn: string]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "file_location = \"datasets/datasets_13996_18858_WA_Fn-UseC_-Telco-Customer-Churn.csv\"\n",
    "file_type = \"csv\"\n",
    "\n",
    "# CSV options\n",
    "infer_schema = \"false\"\n",
    "first_row_is_header = \"True\"\n",
    "delimiter = \",\"\n",
    "\n",
    "# The applied options are for CSV files. For other file types, these will be ignored.\n",
    "df = spark.read.format(file_type) \\\n",
    "  .option(\"inferSchema\", infer_schema) \\\n",
    "  .option(\"header\", first_row_is_header) \\\n",
    "  .option(\"sep\", delimiter) \\\n",
    "  .option(\"nanValue\", ' ') \\\n",
    "  .option(\"nullValue\", ' ') \\\n",
    "  .load(file_location)\n",
    "\n",
    "display(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+-------+----------+------+------------+----------------+---------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------+----------------+--------------------+--------------+------------+-----+\n",
      "|customerID|gender|SeniorCitizen|Partner|Dependents|tenure|PhoneService|   MultipleLines|InternetService|     OnlineSecurity|       OnlineBackup|   DeviceProtection|        TechSupport|        StreamingTV|    StreamingMovies|      Contract|PaperlessBilling|       PaymentMethod|MonthlyCharges|TotalCharges|Churn|\n",
      "+----------+------+-------------+-------+----------+------+------------+----------------+---------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------+----------------+--------------------+--------------+------------+-----+\n",
      "|7590-VHVEG|Female|            0|    Yes|        No|     1|          No|No phone service|            DSL|                 No|                Yes|                 No|                 No|                 No|                 No|Month-to-month|             Yes|    Electronic check|         29.85|       29.85|   No|\n",
      "|5575-GNVDE|  Male|            0|     No|        No|    34|         Yes|              No|            DSL|                Yes|                 No|                Yes|                 No|                 No|                 No|      One year|              No|        Mailed check|         56.95|      1889.5|   No|\n",
      "|3668-QPYBK|  Male|            0|     No|        No|     2|         Yes|              No|            DSL|                Yes|                Yes|                 No|                 No|                 No|                 No|Month-to-month|             Yes|        Mailed check|         53.85|      108.15|  Yes|\n",
      "|7795-CFOCW|  Male|            0|     No|        No|    45|          No|No phone service|            DSL|                Yes|                 No|                Yes|                Yes|                 No|                 No|      One year|              No|Bank transfer (au...|          42.3|     1840.75|   No|\n",
      "|9237-HQITU|Female|            0|     No|        No|     2|         Yes|              No|    Fiber optic|                 No|                 No|                 No|                 No|                 No|                 No|Month-to-month|             Yes|    Electronic check|          70.7|      151.65|  Yes|\n",
      "|9305-CDSKC|Female|            0|     No|        No|     8|         Yes|             Yes|    Fiber optic|                 No|                 No|                Yes|                 No|                Yes|                Yes|Month-to-month|             Yes|    Electronic check|         99.65|       820.5|  Yes|\n",
      "|1452-KIOVK|  Male|            0|     No|       Yes|    22|         Yes|             Yes|    Fiber optic|                 No|                Yes|                 No|                 No|                Yes|                 No|Month-to-month|             Yes|Credit card (auto...|          89.1|      1949.4|   No|\n",
      "|6713-OKOMC|Female|            0|     No|        No|    10|          No|No phone service|            DSL|                Yes|                 No|                 No|                 No|                 No|                 No|Month-to-month|              No|        Mailed check|         29.75|       301.9|   No|\n",
      "|7892-POOKP|Female|            0|    Yes|        No|    28|         Yes|             Yes|    Fiber optic|                 No|                 No|                Yes|                Yes|                Yes|                Yes|Month-to-month|             Yes|    Electronic check|         104.8|     3046.05|  Yes|\n",
      "|6388-TABGU|  Male|            0|     No|       Yes|    62|         Yes|              No|            DSL|                Yes|                Yes|                 No|                 No|                 No|                 No|      One year|              No|Bank transfer (au...|         56.15|     3487.95|   No|\n",
      "|9763-GRSKD|  Male|            0|    Yes|       Yes|    13|         Yes|              No|            DSL|                Yes|                 No|                 No|                 No|                 No|                 No|Month-to-month|             Yes|        Mailed check|         49.95|      587.45|   No|\n",
      "|7469-LKBCI|  Male|            0|     No|        No|    16|         Yes|              No|             No|No internet service|No internet service|No internet service|No internet service|No internet service|No internet service|      Two year|              No|Credit card (auto...|         18.95|       326.8|   No|\n",
      "|8091-TTVAX|  Male|            0|    Yes|        No|    58|         Yes|             Yes|    Fiber optic|                 No|                 No|                Yes|                 No|                Yes|                Yes|      One year|              No|Credit card (auto...|        100.35|      5681.1|   No|\n",
      "|0280-XJGEX|  Male|            0|     No|        No|    49|         Yes|             Yes|    Fiber optic|                 No|                Yes|                Yes|                 No|                Yes|                Yes|Month-to-month|             Yes|Bank transfer (au...|         103.7|      5036.3|  Yes|\n",
      "|5129-JLPIS|  Male|            0|     No|        No|    25|         Yes|              No|    Fiber optic|                Yes|                 No|                Yes|                Yes|                Yes|                Yes|Month-to-month|             Yes|    Electronic check|         105.5|     2686.05|   No|\n",
      "|3655-SNQYZ|Female|            0|    Yes|       Yes|    69|         Yes|             Yes|    Fiber optic|                Yes|                Yes|                Yes|                Yes|                Yes|                Yes|      Two year|              No|Credit card (auto...|        113.25|     7895.15|   No|\n",
      "|8191-XWSZG|Female|            0|     No|        No|    52|         Yes|              No|             No|No internet service|No internet service|No internet service|No internet service|No internet service|No internet service|      One year|              No|        Mailed check|         20.65|     1022.95|   No|\n",
      "|9959-WOFKT|  Male|            0|     No|       Yes|    71|         Yes|             Yes|    Fiber optic|                Yes|                 No|                Yes|                 No|                Yes|                Yes|      Two year|              No|Bank transfer (au...|         106.7|     7382.25|   No|\n",
      "|4190-MFLUW|Female|            0|    Yes|       Yes|    10|         Yes|              No|            DSL|                 No|                 No|                Yes|                Yes|                 No|                 No|Month-to-month|              No|Credit card (auto...|          55.2|      528.35|  Yes|\n",
      "|4183-MYFRB|Female|            0|     No|        No|    21|         Yes|              No|    Fiber optic|                 No|                Yes|                Yes|                 No|                 No|                Yes|Month-to-month|             Yes|    Electronic check|         90.05|      1862.9|   No|\n",
      "+----------+------+-------------+-------+----------+------+------------+----------------+---------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------+----------------+--------------------+--------------+------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+\n",
      "|customerID|gender|SeniorCitizen|Partner|Dependents|tenure|PhoneService|MultipleLines|InternetService|OnlineSecurity|OnlineBackup|DeviceProtection|TechSupport|StreamingTV|StreamingMovies|Contract|PaperlessBilling|PaymentMethod|MonthlyCharges|TotalCharges|Churn|\n",
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+\n",
      "|         0|     0|            0|      0|         0|     0|           0|            0|              0|             0|           0|               0|          0|          0|              0|       0|               0|            0|             0|          11|    0|\n",
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import isnan, when, count, col\n",
    "df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|Churn|count|\n",
      "+-----+-----+\n",
      "|   No| 5174|\n",
      "|  Yes| 1869|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"Churn\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|tenure|\n",
      "+------+\n",
      "|     1|\n",
      "|    34|\n",
      "|     2|\n",
      "|    45|\n",
      "|     2|\n",
      "|     8|\n",
      "|    22|\n",
      "|    10|\n",
      "|    28|\n",
      "|    62|\n",
      "|    13|\n",
      "|    16|\n",
      "|    58|\n",
      "|    49|\n",
      "|    25|\n",
      "|    69|\n",
      "|    52|\n",
      "|    71|\n",
      "|    10|\n",
      "|    21|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"tenure\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+------------------+------------------+\n",
      "|summary|            tenure|      TotalCharges|    MonthlyCharges|\n",
      "+-------+------------------+------------------+------------------+\n",
      "|  count|              7043|              7032|              7043|\n",
      "|   mean| 32.37114865824223|2283.3004408418697| 64.76169246059922|\n",
      "| stddev|24.559481023094442| 2266.771361883145|30.090047097678482|\n",
      "|    min|                 0|             100.2|               100|\n",
      "|    max|                 9|             999.9|             99.95|\n",
      "+-------+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"tenure\", \"TotalCharges\", \"MonthlyCharges\").describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------+----+-----------+----+\n",
      "|SeniorCitizen_InternetService| DSL|Fiber optic|  No|\n",
      "+-----------------------------+----+-----------+----+\n",
      "|                            1| 259|        831|  52|\n",
      "|                            0|2162|       2265|1474|\n",
      "+-----------------------------+----+-----------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.stat.crosstab(\"SeniorCitizen\", \"InternetService\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['customerID',\n",
       " 'gender',\n",
       " 'SeniorCitizen',\n",
       " 'Partner',\n",
       " 'Dependents',\n",
       " 'tenure',\n",
       " 'PhoneService',\n",
       " 'MultipleLines',\n",
       " 'InternetService',\n",
       " 'OnlineSecurity',\n",
       " 'OnlineBackup',\n",
       " 'DeviceProtection',\n",
       " 'TechSupport',\n",
       " 'StreamingTV',\n",
       " 'StreamingMovies',\n",
       " 'Contract',\n",
       " 'PaperlessBilling',\n",
       " 'PaymentMethod',\n",
       " 'MonthlyCharges',\n",
       " 'TotalCharges',\n",
       " 'Churn']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(PhoneService_freqItems=['Yes'], InternetService_freqItems=['Fiber optic'], MultipleLines_freqItems=['Yes'], OnlineSecurity_freqItems=['No'], OnlineBackup_freqItems=['Yes'], DeviceProtection_freqItems=['No'], TechSupport_freqItems=['No'], StreamingTV_freqItems=['Yes'], StreamingMovies_freqItems=['No'])]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.stat.freqItems([\"PhoneService\", \"InternetService\", \"MultipleLines\", \"OnlineSecurity\", \n",
    "                   \"OnlineBackup\", \"DeviceProtection\", \"TechSupport\", \"StreamingTV\", \"StreamingMovies\"], 0.6).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"TotalCharges\", df[\"TotalCharges\"].cast(\"float\"))\n",
    "df = df.withColumn(\"MonthlyCharges\", df[\"MonthlyCharges\"].cast(\"float\"))\n",
    "df = df.withColumn(\"tenure\", df[\"tenure\"].cast(\"int\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Records for training: 4942\n",
      "Records for testing: 2101\n"
     ]
    }
   ],
   "source": [
    "churn_df = df\n",
    "(train_data, test_data) = churn_df.randomSplit([0.7, 0.3], 24)\n",
    "\n",
    "print(f\"Records for training: {str(train_data.count())}\")\n",
    "print(f\"Records for testing: {str(test_data.count())}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import pipeline\n",
    "from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "cat_columns = [\"gender\", \"SeniorCitizen\", \"Partner\", \"Dependents\", \"PhoneService\", \"MultipleLines\", \"InternetService\", \n",
    "               \"OnlineSecurity\", \"OnlineBackup\", \"DeviceProtection\", \"TechSupport\", \"StreamingTV\", \"StreamingMovies\",\n",
    "              \"Contract\", \"PaperlessBilling\", \"PaymentMethod\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"TotalCharges\", df[\"TotalCharges\"].cast(\"float\"))\n",
    "df = df.withColumn(\"MonthlyCharges\", df[\"MonthlyCharges\"].cast(\"float\"))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "16"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(cat_columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "genderindex\n",
      "SeniorCitizenindex\n",
      "Partnerindex\n",
      "Dependentsindex\n",
      "PhoneServiceindex\n",
      "MultipleLinesindex\n",
      "InternetServiceindex\n",
      "OnlineSecurityindex\n",
      "OnlineBackupindex\n",
      "DeviceProtectionindex\n",
      "TechSupportindex\n",
      "StreamingTVindex\n",
      "StreamingMoviesindex\n",
      "Contractindex\n",
      "PaperlessBillingindex\n",
      "PaymentMethodindex\n"
     ]
    }
   ],
   "source": [
    "stage = []\n",
    "for i in cat_columns:\n",
    "    string_indexer = StringIndexer(inputCol=i, outputCol=i + \"index\")\n",
    "    print(string_indexer.getOutputCol())\n",
    "    one_hot = OneHotEncoder(inputCols=[string_indexer.getOutputCol()],\n",
    "                        outputCols=[i + \"catVec\"])\n",
    "    stage += [string_indexer, one_hot]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Imputer\n",
    "imputer = Imputer(inputCols=[\"TotalCharges\"], outputCols=[\"Out_TotalCharges\"])\n",
    "stage += [imputer]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "label_idx = StringIndexer(inputCol=\"Churn\", outputCol=\"label\")\n",
    "stage += [label_idx]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[StringIndexer_a89b3accba7e,\n",
       " OneHotEncoder_fbfd737c41a9,\n",
       " StringIndexer_927d65535359,\n",
       " OneHotEncoder_1e4aef4b5415,\n",
       " StringIndexer_8c9dba0294fc,\n",
       " OneHotEncoder_c3a6628ac928,\n",
       " StringIndexer_fda9336a1461,\n",
       " OneHotEncoder_a129ef515604,\n",
       " StringIndexer_59f1a09127f0,\n",
       " OneHotEncoder_e5d85d73de43,\n",
       " StringIndexer_764880f064a6,\n",
       " OneHotEncoder_e98ca3123f3a,\n",
       " StringIndexer_d66c24be3835,\n",
       " OneHotEncoder_7e466340ac5d,\n",
       " StringIndexer_85025c255102,\n",
       " OneHotEncoder_ce188d9fa2e4,\n",
       " StringIndexer_b490a1773dc7,\n",
       " OneHotEncoder_5a7716a5c16c,\n",
       " StringIndexer_848c72a4bf6f,\n",
       " OneHotEncoder_564139078285,\n",
       " StringIndexer_09ff5eb99813,\n",
       " OneHotEncoder_f887ab1ff4f8,\n",
       " StringIndexer_8bf743934276,\n",
       " OneHotEncoder_3a87170a7fc2,\n",
       " StringIndexer_b117fbb1c2bf,\n",
       " OneHotEncoder_2cf69133ffa5,\n",
       " StringIndexer_93742915bebe,\n",
       " OneHotEncoder_b7d865625929,\n",
       " StringIndexer_faa462f39493,\n",
       " OneHotEncoder_0b564380877b,\n",
       " StringIndexer_399a91df7ae5,\n",
       " OneHotEncoder_a1cf63b8309e,\n",
       " Imputer_a731b973d26c,\n",
       " StringIndexer_fabdfdfaba8c]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+-----+\n",
      "|customerID|gender|SeniorCitizen|Partner|Dependents|tenure|PhoneService|MultipleLines|InternetService|OnlineSecurity|OnlineBackup|DeviceProtection|TechSupport|StreamingTV|StreamingMovies|Contract|PaperlessBilling|PaymentMethod|MonthlyCharges|TotalCharges|Churn|label|\n",
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+-----+\n",
      "|0002-ORFBO|Female|            0|    Yes|       Yes|     9|         Yes|           No|            DSL|            No|         Yes|              No|        Yes|        Yes|             No|One year|             Yes| Mailed check|          65.6|       593.3|   No|  0.0|\n",
      "+----------+------+-------------+-------+----------+------+------------+-------------+---------------+--------------+------------+----------------+-----------+-----------+---------------+--------+----------------+-------------+--------------+------------+-----+-----+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "temp = label_idx.fit(train_data).transform(train_data)\n",
    "temp.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|MonthlyCharges|\n",
      "+--------------+\n",
      "|         29.85|\n",
      "|         56.95|\n",
      "|         53.85|\n",
      "|          42.3|\n",
      "|          70.7|\n",
      "|         99.65|\n",
      "|          89.1|\n",
      "|         29.75|\n",
      "|         104.8|\n",
      "|         56.15|\n",
      "|         49.95|\n",
      "|         18.95|\n",
      "|        100.35|\n",
      "|         103.7|\n",
      "|         105.5|\n",
      "|        113.25|\n",
      "|         20.65|\n",
      "|         106.7|\n",
      "|          55.2|\n",
      "|         90.05|\n",
      "+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"MonthlyCharges\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|TotalCharges|\n",
      "+------------+\n",
      "|       29.85|\n",
      "|      1889.5|\n",
      "|      108.15|\n",
      "|     1840.75|\n",
      "|      151.65|\n",
      "|       820.5|\n",
      "|      1949.4|\n",
      "|       301.9|\n",
      "|     3046.05|\n",
      "|     3487.95|\n",
      "|      587.45|\n",
      "|       326.8|\n",
      "|      5681.1|\n",
      "|      5036.3|\n",
      "|     2686.05|\n",
      "|     7895.15|\n",
      "|     1022.95|\n",
      "|     7382.25|\n",
      "|      528.35|\n",
      "|      1862.9|\n",
      "+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"TotalCharges\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- SeniorCitizen: string (nullable = true)\n",
      " |-- Partner: string (nullable = true)\n",
      " |-- Dependents: string (nullable = true)\n",
      " |-- tenure: integer (nullable = true)\n",
      " |-- PhoneService: string (nullable = true)\n",
      " |-- MultipleLines: string (nullable = true)\n",
      " |-- InternetService: string (nullable = true)\n",
      " |-- OnlineSecurity: string (nullable = true)\n",
      " |-- OnlineBackup: string (nullable = true)\n",
      " |-- DeviceProtection: string (nullable = true)\n",
      " |-- TechSupport: string (nullable = true)\n",
      " |-- StreamingTV: string (nullable = true)\n",
      " |-- StreamingMovies: string (nullable = true)\n",
      " |-- Contract: string (nullable = true)\n",
      " |-- PaperlessBilling: string (nullable = true)\n",
      " |-- PaymentMethod: string (nullable = true)\n",
      " |-- MonthlyCharges: float (nullable = true)\n",
      " |-- TotalCharges: float (nullable = true)\n",
      " |-- Churn: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"TotalCharges\", df[\"TotalCharges\"].cast(\"float\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\"MonthlyCharges\", df[\"MonthlyCharges\"].cast(\"float\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- SeniorCitizen: string (nullable = true)\n",
      " |-- Partner: string (nullable = true)\n",
      " |-- Dependents: string (nullable = true)\n",
      " |-- tenure: integer (nullable = true)\n",
      " |-- PhoneService: string (nullable = true)\n",
      " |-- MultipleLines: string (nullable = true)\n",
      " |-- InternetService: string (nullable = true)\n",
      " |-- OnlineSecurity: string (nullable = true)\n",
      " |-- OnlineBackup: string (nullable = true)\n",
      " |-- DeviceProtection: string (nullable = true)\n",
      " |-- TechSupport: string (nullable = true)\n",
      " |-- StreamingTV: string (nullable = true)\n",
      " |-- StreamingMovies: string (nullable = true)\n",
      " |-- Contract: string (nullable = true)\n",
      " |-- PaperlessBilling: string (nullable = true)\n",
      " |-- PaymentMethod: string (nullable = true)\n",
      " |-- MonthlyCharges: float (nullable = true)\n",
      " |-- TotalCharges: float (nullable = true)\n",
      " |-- Churn: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.6511738319154315"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.stat.corr(\"TotalCharges\", \"MonthlyCharges\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import QuantileDiscretizer\n",
    "tenure_bin = QuantileDiscretizer(numBuckets=3, inputCol=\"tenure\", outputCol=\"tenure_bin\")\n",
    "stage += [tenure_bin]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[StringIndexer_a89b3accba7e,\n",
       " OneHotEncoder_fbfd737c41a9,\n",
       " StringIndexer_927d65535359,\n",
       " OneHotEncoder_1e4aef4b5415,\n",
       " StringIndexer_8c9dba0294fc,\n",
       " OneHotEncoder_c3a6628ac928,\n",
       " StringIndexer_fda9336a1461,\n",
       " OneHotEncoder_a129ef515604,\n",
       " StringIndexer_59f1a09127f0,\n",
       " OneHotEncoder_e5d85d73de43,\n",
       " StringIndexer_764880f064a6,\n",
       " OneHotEncoder_e98ca3123f3a,\n",
       " StringIndexer_d66c24be3835,\n",
       " OneHotEncoder_7e466340ac5d,\n",
       " StringIndexer_85025c255102,\n",
       " OneHotEncoder_ce188d9fa2e4,\n",
       " StringIndexer_b490a1773dc7,\n",
       " OneHotEncoder_5a7716a5c16c,\n",
       " StringIndexer_848c72a4bf6f,\n",
       " OneHotEncoder_564139078285,\n",
       " StringIndexer_09ff5eb99813,\n",
       " OneHotEncoder_f887ab1ff4f8,\n",
       " StringIndexer_8bf743934276,\n",
       " OneHotEncoder_3a87170a7fc2,\n",
       " StringIndexer_b117fbb1c2bf,\n",
       " OneHotEncoder_2cf69133ffa5,\n",
       " StringIndexer_93742915bebe,\n",
       " OneHotEncoder_b7d865625929,\n",
       " StringIndexer_faa462f39493,\n",
       " OneHotEncoder_0b564380877b,\n",
       " StringIndexer_399a91df7ae5,\n",
       " OneHotEncoder_a1cf63b8309e,\n",
       " Imputer_a731b973d26c,\n",
       " StringIndexer_fabdfdfaba8c,\n",
       " QuantileDiscretizer_7a138cf41e2f]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "numeric_cols = [\"tenure_bin\", \"Out_TotalCharges\", \"MonthlyCharges\"]\n",
    "assembleInputs = assemblerInputs = [c + \"catVec\" for c in cat_columns] + numeric_cols\n",
    "assembler = VectorAssembler(inputCols=assembleInputs, outputCol=\"feature\")\n",
    "stage += [assembler]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[StringIndexer_a89b3accba7e,\n",
       " OneHotEncoder_fbfd737c41a9,\n",
       " StringIndexer_927d65535359,\n",
       " OneHotEncoder_1e4aef4b5415,\n",
       " StringIndexer_8c9dba0294fc,\n",
       " OneHotEncoder_c3a6628ac928,\n",
       " StringIndexer_fda9336a1461,\n",
       " OneHotEncoder_a129ef515604,\n",
       " StringIndexer_59f1a09127f0,\n",
       " OneHotEncoder_e5d85d73de43,\n",
       " StringIndexer_764880f064a6,\n",
       " OneHotEncoder_e98ca3123f3a,\n",
       " StringIndexer_d66c24be3835,\n",
       " OneHotEncoder_7e466340ac5d,\n",
       " StringIndexer_85025c255102,\n",
       " OneHotEncoder_ce188d9fa2e4,\n",
       " StringIndexer_b490a1773dc7,\n",
       " OneHotEncoder_5a7716a5c16c,\n",
       " StringIndexer_848c72a4bf6f,\n",
       " OneHotEncoder_564139078285,\n",
       " StringIndexer_09ff5eb99813,\n",
       " OneHotEncoder_f887ab1ff4f8,\n",
       " StringIndexer_8bf743934276,\n",
       " OneHotEncoder_3a87170a7fc2,\n",
       " StringIndexer_b117fbb1c2bf,\n",
       " OneHotEncoder_2cf69133ffa5,\n",
       " StringIndexer_93742915bebe,\n",
       " OneHotEncoder_b7d865625929,\n",
       " StringIndexer_faa462f39493,\n",
       " OneHotEncoder_0b564380877b,\n",
       " StringIndexer_399a91df7ae5,\n",
       " OneHotEncoder_a1cf63b8309e,\n",
       " Imputer_a731b973d26c,\n",
       " StringIndexer_fabdfdfaba8c,\n",
       " QuantileDiscretizer_7a138cf41e2f,\n",
       " VectorAssembler_b9421cb13cff]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = Pipeline().setStages(stage)\n",
    "pipelineModel = pipeline.fit(train_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df = pipelineModel.transform(train_data)\n",
    "test_df = pipelineModel.transform(test_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "57"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(train_df.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(customerID='0002-ORFBO', gender='Female', SeniorCitizen='0', Partner='Yes', Dependents='Yes', tenure=9, PhoneService='Yes', MultipleLines='No', InternetService='DSL', OnlineSecurity='No', OnlineBackup='Yes', DeviceProtection='No', TechSupport='Yes', StreamingTV='Yes', StreamingMovies='No', Contract='One year', PaperlessBilling='Yes', PaymentMethod='Mailed check', MonthlyCharges=65.5999984741211, TotalCharges=593.2999877929688, Churn='No', genderindex=1.0, gendercatVec=SparseVector(1, {}), SeniorCitizenindex=0.0, SeniorCitizencatVec=SparseVector(1, {0: 1.0}), Partnerindex=1.0, PartnercatVec=SparseVector(1, {}), Dependentsindex=1.0, DependentscatVec=SparseVector(1, {}), PhoneServiceindex=0.0, PhoneServicecatVec=SparseVector(1, {0: 1.0}), MultipleLinesindex=0.0, MultipleLinescatVec=SparseVector(2, {0: 1.0}), InternetServiceindex=1.0, InternetServicecatVec=SparseVector(2, {1: 1.0}), OnlineSecurityindex=0.0, OnlineSecuritycatVec=SparseVector(2, {0: 1.0}), OnlineBackupindex=1.0, OnlineBackupcatVec=SparseVector(2, {1: 1.0}), DeviceProtectionindex=0.0, DeviceProtectioncatVec=SparseVector(2, {0: 1.0}), TechSupportindex=1.0, TechSupportcatVec=SparseVector(2, {1: 1.0}), StreamingTVindex=1.0, StreamingTVcatVec=SparseVector(2, {1: 1.0}), StreamingMoviesindex=0.0, StreamingMoviescatVec=SparseVector(2, {0: 1.0}), Contractindex=2.0, ContractcatVec=SparseVector(2, {}), PaperlessBillingindex=0.0, PaperlessBillingcatVec=SparseVector(1, {0: 1.0}), PaymentMethodindex=1.0, PaymentMethodcatVec=SparseVector(3, {1: 1.0}), Out_TotalCharges=593.2999877929688, label=0.0, tenure_bin=0.0, feature=SparseVector(30, {1: 1.0, 4: 1.0, 5: 1.0, 8: 1.0, 9: 1.0, 12: 1.0, 13: 1.0, 16: 1.0, 18: 1.0, 19: 1.0, 23: 1.0, 25: 1.0, 28: 593.3, 29: 65.6}))]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_df.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|tenure_bin|\n",
      "+----------+\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       2.0|\n",
      "|       0.0|\n",
      "|       2.0|\n",
      "|       2.0|\n",
      "|       0.0|\n",
      "|       2.0|\n",
      "|       1.0|\n",
      "|       0.0|\n",
      "|       1.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       2.0|\n",
      "|       2.0|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train_df.select(\"tenure_bin\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "lr = LogisticRegression(labelCol=\"label\", featuresCol=\"feature\", maxIter=10)\n",
    "lrModel = lr.fit(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "coefficients: [0.034280961208810704,-0.3312162846190503,-0.10403353386750051,0.12868949858748596,-0.6191319042742331,-0.24717797905720307,0.02685652018272491,0.5947672944599564,-0.43247345944430066,0.26034309970076586,-0.07873494210934841,0.17599225888273645,0.025094379273215246,0.13427108952998187,0.07038245400016194,0.2474620870056942,-0.06197276941501819,-0.03742781503932004,0.2449855258719482,0.012591165496948949,0.19287810202563474,0.7074105781395923,-0.852599383222226,0.2978156779228366,0.3370045996985892,-0.08658398635534728,0.010002029823934406,-0.7221850206119766,-0.00011665000400444835,0.005179321118112393]\n",
      "Intercept: -1.1540815262190598\n"
     ]
    }
   ],
   "source": [
    "print(f\"coefficients: {str(lrModel.coefficients)}\")\n",
    "print(f\"Intercept: {str(lrModel.intercept)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "summary = lrModel.summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "accuracy: 0.8057466612707406, FP: 0.3625173099768732, TP: 0.8057466612707406, fmeasure: 0.7993759299320431, precision: 0.7971094940545889, Recall: 0.8057466612707406\n"
     ]
    }
   ],
   "source": [
    "print(f\"accuracy: {summary.accuracy}, FP: {summary.weightedFalsePositiveRate}, TP: {summary.weightedTruePositiveRate}, fmeasure: {summary.weightedFMeasure()}, precision: {summary.weightedPrecision}, Recall: {summary.weightedRecall}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "LogisticRegressionModel: uid=LogisticRegression_e206a36e2618, numClasses=2, numFeatures=30"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[customerID: string, gender: string, SeniorCitizen: string, Partner: string, Dependents: string, tenure: int, PhoneService: string, MultipleLines: string, InternetService: string, OnlineSecurity: string, OnlineBackup: string, DeviceProtection: string, TechSupport: string, StreamingTV: string, StreamingMovies: string, Contract: string, PaperlessBilling: string, PaymentMethod: string, MonthlyCharges: float, TotalCharges: float, Churn: string, genderindex: double, gendercatVec: vector, SeniorCitizenindex: double, SeniorCitizencatVec: vector, Partnerindex: double, PartnercatVec: vector, Dependentsindex: double, DependentscatVec: vector, PhoneServiceindex: double, PhoneServicecatVec: vector, MultipleLinesindex: double, MultipleLinescatVec: vector, InternetServiceindex: double, InternetServicecatVec: vector, OnlineSecurityindex: double, OnlineSecuritycatVec: vector, OnlineBackupindex: double, OnlineBackupcatVec: vector, DeviceProtectionindex: double, DeviceProtectioncatVec: vector, TechSupportindex: double, TechSupportcatVec: vector, StreamingTVindex: double, StreamingTVcatVec: vector, StreamingMoviesindex: double, StreamingMoviescatVec: vector, Contractindex: double, ContractcatVec: vector, PaperlessBillingindex: double, PaperlessBillingcatVec: vector, PaymentMethodindex: double, PaymentMethodcatVec: vector, Out_TotalCharges: float, label: double, tenure_bin: double, feature: vector]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "'ROC'"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(lrModel, train_df, \"ROC\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "predictions = lrModel.transform(test_df)\n",
    "evaluate_model = BinaryClassificationEvaluator(rawPredictionCol=\"prediction\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Area under curve: 0.7041164527040232\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'areaUnderROC'"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "area_under_curve = evaluate_model.evaluate(predictions)\n",
    "print(f\"Area under curve: {area_under_curve}\")\n",
    "evaluate_model.getMetricName()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
    "# results = predictions.select([\"prediction\", \"label\"])\n",
    "# results_collect = results.collect()\n",
    "# results_list = [(float(i[0]), float(i[1])) for i in results_collect]\n",
    "# prediction_and_label = spark.parallelize(results_list)\n",
    "# metrics = BinaryClassificationMetrics(prediction_and_label)\n",
    "\n",
    "# print(f\"Area under PR: {metrics.areaUnderPR}, Area under ROC {metrics.areaUnderROC}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|Churn|count|\n",
      "+-----+-----+\n",
      "|   No| 1534|\n",
      "|  Yes|  567|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_df.groupBy(\"Churn\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "count = predictions.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2101"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "correct = results.filter(results.prediction == results.label ).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1652"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "correct"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "298"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tp = results.filter(results.prediction == 1.0).filter(results.prediction == results.label).count()\n",
    "tp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "180"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fp = results.filter(results.prediction == 1.0).filter(results.prediction != results.label).count()\n",
    "fp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "269"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fn = results.filter(results.prediction == 0.0).filter(results.prediction != results.label).count()\n",
    "fn "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1354"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tn = results.filter(results.prediction == 0.0).filter(results.prediction == results.label).count()\n",
    "tn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.786292241789624"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "accuracy = (tp+tn)/count\n",
    "accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.6234309623430963"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "precision = tp/(tp+fp)\n",
    "precision"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.5255731922398589"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "recall = tp/(tp+fn)\n",
    "recall"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\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, [5, 10, 20])\n",
    "            .build())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "cv = CrossValidator(estimator=lr, estimatorParamMaps=paramgrid, evaluator=evaluate_model, numFolds=5)\n",
    "cv_model = cv.fit(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions = cv_model.bestModel.transform(test_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.7007811188602149"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluate_model.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"estimator: estimator to be cross-validated (current: LogisticRegression_e206a36e2618)\\nestimatorParamMaps: estimator param maps (current: [{Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.01, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 0.5, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 5}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 10}, {Param(parent='LogisticRegression_e206a36e2618', name='regParam', doc='regularization parameter (>= 0).'): 2.0, Param(parent='LogisticRegression_e206a36e2618', name='elasticNetParam', doc='the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty.'): 1.0, Param(parent='LogisticRegression_e206a36e2618', name='maxIter', doc='max number of iterations (>= 0).'): 20}])\\nevaluator: evaluator used to select hyper-parameters that maximize the validator metric (current: BinaryClassificationEvaluator_c94c50eecffd)\\nnumFolds: number of folds for cross validation (default: 3, current: 5)\\nseed: random seed. (default: 7959892220179297753)\""
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cv_model.explainParams()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "rf = RandomForestClassifier(labelCol=\"label\", featuresCol=\"feature\").setImpurity(\"gini\").setMaxDepth(6).setNumTrees(50).setFeatureSubsetStrategy(\"auto\").setSeed(1010)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "rf_model = rf.fit(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions = rf_model.transform(test_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}