Spam_Classification_with_Spark.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kt4dQPvEVEHm"
      },
      "source": []
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "zBGR8Bv8PMtp"
      },
      "outputs": [],
      "source": [
        "# innstall java\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "id": "4EikmM9nPPdZ"
      },
      "outputs": [],
      "source": [
        "# install spark (change the version number if needed)\n",
        "!wget -q https://archive.apache.org/dist/spark/spark-3.0.0/spark-3.0.0-bin-hadoop3.2.tgz"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "5ApuuTo9RAl4"
      },
      "outputs": [],
      "source": [
        "# unzip the spark file to the current folder\n",
        "!tar xf spark-3.0.0-bin-hadoop3.2.tgz"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "id": "wL7FKWh-RAjg"
      },
      "outputs": [],
      "source": [
        "# set your spark folder to your system path environment. \n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.0.0-bin-hadoop3.2\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "lJIwWZ6wRAgs"
      },
      "outputs": [],
      "source": [
        "# install findspark using pip\n",
        "!pip install -q findspark"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "qOiagWo9RAeL"
      },
      "outputs": [],
      "source": []
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "id": "A_GvJiBMPPbC"
      },
      "outputs": [],
      "source": [
        "import findspark"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "id": "hkiWD9sxPPYw"
      },
      "outputs": [],
      "source": [
        "findspark.init()\n",
        "from pyspark.sql import SparkSession"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "id": "HOQHyHV-PPWo"
      },
      "outputs": [],
      "source": [
        "spark = SparkSession.builder.appName('SMSSpamCollection').getOrCreate()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NJ51pSpGz1f1",
        "outputId": "010e6037-3af4-42d5-9a76-d928411638f2"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "root\n",
            " |-- class: string (nullable = true)\n",
            " |-- text: string (nullable = true)\n",
            "\n"
          ]
        }
      ],
      "source": [
        "df = spark.read.options(inferSchema='True',delimiter='\\t').csv(\"SMSSpamCollection.csv\")\n",
        "# df = spark.read.options(inferSchema='True', ).csv(\"spam.csv\")\n",
        "\n",
        "## Rename the columns\n",
        "df = df.withColumnRenamed(\"_c0\", \"class\").withColumnRenamed(\"_c1\", \"text\")\n",
        "# df = df.withColumnRenamed(\"_v1\", \"class\").withColumnRenamed(\"_v2\", \"text\")\n",
        "df.printSchema()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {
        "id": "oDZ9eUmB2WMH"
      },
      "outputs": [],
      "source": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1uwg52j74AF0"
      },
      "source": [
        "### Create a new length feature (new column w/ the length of the text)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "uDcWlmPK1F2U",
        "outputId": "cb380229-dc70-4ec5-9200-05f128e0fcb1"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-----+--------------------+----------+\n",
            "|class|                text|textLength|\n",
            "+-----+--------------------+----------+\n",
            "|  ham|Go until jurong p...|       111|\n",
            "|  ham|Ok lar... Joking ...|        29|\n",
            "| spam|Free entry in 2 a...|       155|\n",
            "|  ham|U dun say so earl...|        49|\n",
            "|  ham|Nah I don't think...|        61|\n",
            "| spam|FreeMsg Hey there...|       147|\n",
            "|  ham|Even my brother i...|        77|\n",
            "|  ham|As per your reque...|       160|\n",
            "| spam|WINNER!! As a val...|       157|\n",
            "| spam|Had your mobile 1...|       154|\n",
            "|  ham|I'm gonna be home...|       109|\n",
            "| spam|SIX chances to wi...|       136|\n",
            "| spam|URGENT! You have ...|       155|\n",
            "|  ham|I've been searchi...|       196|\n",
            "|  ham|I HAVE A DATE ON ...|        35|\n",
            "| spam|XXXMobileMovieClu...|       149|\n",
            "|  ham|Oh k...i'm watchi...|        26|\n",
            "|  ham|Eh u remember how...|        81|\n",
            "|  ham|Fine if thats th...|        56|\n",
            "| spam|England v Macedon...|       155|\n",
            "+-----+--------------------+----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "import pyspark.sql.functions as F\n",
        "\n",
        "df = df.withColumn(\"textLength\", F.length(\"text\"))\n",
        "df.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "S4CzYynV3-0B"
      },
      "source": [
        "### What do you notice ?"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_Bxe_jf66n5B"
      },
      "source": [
        "#### Visual inspection shows that the average length of the spam messages is longer."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "HVLpTb2m1Fy6",
        "outputId": "58f18d06-6374-4a23-de1d-9d1e84376323"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-----+-----------------+\n",
            "|class|  avg(textLength)|\n",
            "+-----+-----------------+\n",
            "|  ham|71.45431945307645|\n",
            "| spam|138.6706827309237|\n",
            "+-----+-----------------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "## We can confirm this below\n",
        "df.groupby('class').mean().show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "SlKd5AzH4Q9C"
      },
      "source": [
        "### Create feature transformers"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "metadata": {
        "id": "79OE9LdS1FnF"
      },
      "outputs": [],
      "source": [
        "from pyspark.ml.feature import Tokenizer, StopWordsRemover, CountVectorizer, IDF, StringIndexer, NGram\n",
        "from pyspark.ml.linalg import Vectors\n",
        "from pyspark.ml.feature import VectorAssembler\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Ar5HB_F39P7H"
      },
      "source": [
        "### Use Pipeline to create a data pre-processing pipeline as follows "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "metadata": {
        "id": "7hka73y81FjV"
      },
      "outputs": [],
      "source": [
        "## Use VectorAssembler to create an assembler of tf_idf feature with length \n",
        "\n",
        "## Change to True to use NGRAM\n",
        "use_ngram = False\n",
        "\n",
        "# Indexing class column to a numeric label \n",
        "data_to_num = StringIndexer(inputCol='class', outputCol='label')\n",
        "\n",
        "# Tokenization\n",
        "tokenizer = Tokenizer(inputCol=\"text\", outputCol=\"token_text\")\n",
        "\n",
        "# # • Ngrams \n",
        "if use_ngram:\n",
        "  ngram = NGram(n=2, inputCol = \"token_text\", outputCol=\"ngrams\")\n",
        "  # • Count vertorization\n",
        "  count_vec = CountVectorizer(inputCol='ngrams', outputCol='count_vec_stop')\n",
        "\n",
        "  # • Stop words removal\n",
        "  # stop_remove = StopWordsRemover(inputCol='ngrams', outputCol='stop_tokens')\n",
        "else:\n",
        "  # • Stop words removal\n",
        "  stop_remove = StopWordsRemover(inputCol='token_text', outputCol='stop_tokens')\n",
        "  # • Count vertorization\n",
        "  count_vec = CountVectorizer(inputCol='stop_tokens', outputCol='count_vec_stop')\n",
        "\n",
        "# • IDF\n",
        "idf = IDF(inputCol=\"count_vec_stop\", outputCol=\"tf_idf\")\n",
        "\n",
        "# • Vector assembling\n",
        "vec_assembler = VectorAssembler(inputCols=['tf_idf', 'textLength'], outputCol='features')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "metadata": {
        "id": "LL9QNUMm1FLF"
      },
      "outputs": [],
      "source": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Kru5OfBT90Ts"
      },
      "source": [
        "### Transform the data DataFrame through the pipeline (last column should be called ‘features’)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "cxI7_ADy1E-2"
      },
      "outputs": [],
      "source": [
        "from pyspark.ml import Pipeline\n",
        "\n",
        "## Use Pipeline to create a data pre-processing pipeline as follows\n",
        "spam_pipe = Pipeline(stages=[data_to_num, tokenizer, stop_remove, count_vec, idf, vec_assembler])\n",
        "\n",
        "## Fit the pipeline to the data\n",
        "spam_cleaner = spam_pipe.fit(df)\n",
        "spam_data_clean = spam_cleaner.transform(df)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "zzXCGdyy1E7f"
      },
      "outputs": [],
      "source": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wfwq4GM8AF19"
      },
      "source": [
        "#### Import NaiveBayse model from pyspark.ml.classification"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 17,
      "metadata": {
        "id": "wdeN6KRX1E3-"
      },
      "outputs": [],
      "source": [
        "from pyspark.ml.classification import NaiveBayes\n",
        "nb = NaiveBayes(featuresCol='features', labelCol='label')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "id": "NF_CyIbs1E1b"
      },
      "outputs": [],
      "source": [
        "# Create a training DataFrame by selection the “label” and “features” column"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 19,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "AOlgwCUy1EwM",
        "outputId": "120023f7-164e-49aa-9325-b2b6c805e646"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-----+--------------------+\n",
            "|label|            features|\n",
            "+-----+--------------------+\n",
            "|  0.0|(13424,[7,11,31,6...|\n",
            "|  0.0|(13424,[0,24,297,...|\n",
            "|  1.0|(13424,[2,13,19,3...|\n",
            "|  0.0|(13424,[0,70,80,1...|\n",
            "|  0.0|(13424,[36,134,31...|\n",
            "|  1.0|(13424,[10,60,139...|\n",
            "|  0.0|(13424,[10,53,103...|\n",
            "|  0.0|(13424,[125,184,4...|\n",
            "|  1.0|(13424,[1,47,118,...|\n",
            "|  1.0|(13424,[0,1,13,27...|\n",
            "|  0.0|(13424,[18,43,120...|\n",
            "|  1.0|(13424,[8,17,37,8...|\n",
            "|  1.0|(13424,[13,30,47,...|\n",
            "|  0.0|(13424,[39,96,217...|\n",
            "|  0.0|(13424,[552,1697,...|\n",
            "|  1.0|(13424,[30,109,11...|\n",
            "|  0.0|(13424,[82,214,47...|\n",
            "|  0.0|(13424,[0,2,49,13...|\n",
            "|  0.0|(13424,[0,74,105,...|\n",
            "|  1.0|(13424,[4,30,33,5...|\n",
            "+-----+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "spam_data_clean = spam_data_clean.select(['label','features'])\n",
        "spam_data_clean.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 20,
      "metadata": {
        "id": "J1IKg5nGA2mB"
      },
      "outputs": [],
      "source": [
        "## Use random split to split the training data\n",
        "(train_data, test_data) = spam_data_clean.randomSplit([0.8, 0.2])\n",
        "\n",
        "## Train your model (fit method)\n",
        "predictor = nb.fit(train_data)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 21,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "W_XlgKKnA2jg",
        "outputId": "0d164f1f-ecce-4974-e127-de34b0e9a8f1"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+-----+--------------------+--------------------+--------------------+----------+\n",
            "|label|            features|       rawPrediction|         probability|prediction|\n",
            "+-----+--------------------+--------------------+--------------------+----------+\n",
            "|  0.0|(13424,[0,1,5,20,...|[-803.72607060426...|[1.0,4.6867109237...|       0.0|\n",
            "|  0.0|(13424,[0,1,7,8,1...|[-1176.6256981431...|[1.0,3.5103454115...|       0.0|\n",
            "|  0.0|(13424,[0,1,7,15,...|[-660.21286574480...|[1.0,8.4035947835...|       0.0|\n",
            "|  0.0|(13424,[0,1,12,33...|[-442.40059174310...|[1.0,8.8635057891...|       0.0|\n",
            "|  0.0|(13424,[0,1,14,31...|[-216.45339120277...|[1.0,1.2320726407...|       0.0|\n",
            "|  0.0|(13424,[0,1,17,19...|[-804.73601390432...|[1.0,5.7980011243...|       0.0|\n",
            "|  0.0|(13424,[0,1,27,35...|[-1481.7694760139...|[0.99999999999996...|       0.0|\n",
            "|  0.0|(13424,[0,1,416,6...|[-301.70765051381...|[1.0,3.5017311139...|       0.0|\n",
            "|  0.0|(13424,[0,2,3,5,3...|[-505.42201187660...|[1.0,6.4238592658...|       0.0|\n",
            "|  0.0|(13424,[0,2,3,6,9...|[-3313.3581494374...|[1.0,2.2394052787...|       0.0|\n",
            "|  0.0|(13424,[0,2,3,8,2...|[-1618.5536471519...|[1.0,1.8679887178...|       0.0|\n",
            "|  0.0|(13424,[0,2,4,5,7...|[-996.08617407984...|[1.0,2.1783189017...|       0.0|\n",
            "|  0.0|(13424,[0,2,4,8,2...|[-1416.6765031732...|[1.0,6.3955898088...|       0.0|\n",
            "|  0.0|(13424,[0,2,4,11,...|[-1135.0399980359...|[1.0,6.3872364093...|       0.0|\n",
            "|  0.0|(13424,[0,2,4,44,...|[-1912.1615172964...|[1.0,2.5650813505...|       0.0|\n",
            "|  0.0|(13424,[0,2,7,8,1...|[-709.22862740811...|[1.0,1.9427229065...|       0.0|\n",
            "|  0.0|(13424,[0,2,7,8,3...|[-1151.4590803279...|[1.0,2.8209150336...|       0.0|\n",
            "|  0.0|(13424,[0,2,7,11,...|[-733.28546357660...|[1.0,8.1240539144...|       0.0|\n",
            "|  0.0|(13424,[0,2,7,11,...|[-1430.5090871769...|[1.0,2.7658829280...|       0.0|\n",
            "|  0.0|(13424,[0,2,7,31,...|[-656.03137750175...|[1.0,2.8658144241...|       0.0|\n",
            "+-----+--------------------+--------------------+--------------------+----------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "## Apply your model to test data\n",
        "test_prediction = predictor.transform(test_data)\n",
        "test_prediction.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3GecoJ27CJn7"
      },
      "source": [
        "### Evaluate accuracy using MulticlassClassificationEvaluator from pyspark.ml.evaluation"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 22,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "bgbblN8CA2en",
        "outputId": "ead0506c-5852-4370-cf3c-7d8d2a3277ae"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Model Accuracy: 0.92\n"
          ]
        }
      ],
      "source": [
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "mcc_eval_acc = MulticlassClassificationEvaluator()\n",
        "model_acc = mcc_eval_acc.evaluate(test_prediction)\n",
        "\n",
        "print(\"Model Accuracy: {:.2f}\".format(model_acc))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 28,
      "metadata": {
        "id": "wTXdprgKOmWG"
      },
      "outputs": [],
      "source": [
        "# save preprocessor\n",
        "spam_cleaner.write().overwrite().save(\"spam_cleaner_pipeline\")\n",
        "\n",
        "# save model\n",
        "predictor.write().overwrite().save(\"spam_classifier_model\")\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 29,
      "metadata": {
        "id": "DBZoDKwwP_ng"
      },
      "outputs": [],
      "source": [
        "# load preprocessor\n",
        "from pyspark.ml import PipelineModel\n",
        "spam_cleaner_loaded = PipelineModel.load(\"spam_cleaner_pipeline\")\n",
        "\n",
        "# load model\n",
        "from pyspark.ml.classification import NaiveBayesModel\n",
        "predictor_loaded = NaiveBayesModel.load(\"spam_classifier_model\")\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 30,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "kkvK_ACrOmTC",
        "outputId": "8c5528bd-6693-43fa-c78f-57653ab1a272"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Spam\n"
          ]
        }
      ],
      "source": [
        "\n",
        "# preprocess the text\n",
        "input_str = \"URGENT! You have won a 1 week FREE membership in our £100,000 Prize Jackpot! Txt the word: CLAIM to No: 81010\"\n",
        "\n",
        "# input_str = \"We would like to invite you and your families to the annual company picnic that will be held on Saturday, June 15th, at the park near the office. The picnic will start at 11 am and end at 3 pm.\"\n",
        "input_df = spark.createDataFrame([(input_str,)], [\"text\"])\n",
        "input_df = input_df.withColumn(\"textLength\", F.length(\"text\"))\n",
        "\n",
        "clean_text = spam_cleaner_loaded.transform(input_df)\n",
        "\n",
        "# predict the class label\n",
        "prediction = predictor_loaded.transform(clean_text)\n",
        "prediction = int(prediction.select(\"prediction\").first()[0])\n",
        "\n",
        "if prediction == 0:\n",
        "  print(\"Not Spam\")\n",
        "else:\n",
        "  print(\"Spam\")\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 25,
      "metadata": {
        "id": "Q8r4S_QaOl8k"
      },
      "outputs": [],
      "source": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NTUg44OOFq_o"
      },
      "source": [
        "## Results\n",
        "#### Model Accuracy was 92% without NGrams\n",
        "#### Model Accuracy was 62% with NGrams\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "base",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "version": "3.8.8"
    },
    "vscode": {
      "interpreter": {
        "hash": "0252c481d34ac53c1ac1678c9fe6fac13794176d753854d881303729516622a7"
      }
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}