Spam_detection_filter.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init('/home/bowen/spark-2.4.4-bin-hadoop2.7/')\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('nlp_spam_detection').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv('SMSSpamCollection', inferSchema=True, sep='\\t')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      " |-- _c1: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumnRenamed('_c0', 'label')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumnRenamed('_c1', 'text')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- label: string (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      " |-- length: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import CountVectorizer, Tokenizer, StopWordsRemover, IDF, StringIndexer, VectorAssembler\n",
    "from pyspark.sql.functions import length "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('length', length(df['text']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----------------+\n",
      "|label|      avg(length)|\n",
      "+-----+-----------------+\n",
      "|  ham|71.45431945307645|\n",
      "| spam|138.6706827309237|\n",
      "+-----+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('label').mean().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenizer = Tokenizer(inputCol='text', outputCol='token_text')\n",
    "stop_remover = StopWordsRemover(inputCol='token_text', outputCol='stop_token')\n",
    "cv = CountVectorizer(inputCol='stop_token', outputCol='count_vec')\n",
    "idf = IDF(inputCol='count_vec',outputCol='tf_idf')\n",
    "label_numeric = StringIndexer(inputCol='label', outputCol='label_num')\n",
    "assembler = VectorAssembler(inputCols=['tf_idf', 'length'], outputCol='features')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import NaiveBayes\n",
    "nb = NaiveBayes(labelCol='label_num')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "#use pipeline\n",
    "from pyspark.ml import Pipeline\n",
    "pipeline = Pipeline(stages=[label_numeric, tokenizer, stop_remover, cv, idf, assembler])\n",
    "cleaner = pipeline.fit(df)\n",
    "final_df = cleaner.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(label_num=0.0, features=SparseVector(13424, {7: 3.1126, 11: 3.2055, 31: 3.822, 61: 4.2072, 72: 4.322, 344: 5.4072, 625: 5.918, 731: 6.1411, 1409: 6.6801, 1598: 6.8343, 4485: 7.5274, 6440: 7.9329, 8092: 7.9329, 8838: 7.9329, 11344: 7.9329, 12979: 7.9329, 13423: 111.0}))]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dataset = final_df.select(['label_num', 'features'])\n",
    "dataset.head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_set, test_set = dataset.randomSplit([0.8, 0.2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "#use Bayes model\n",
    "spam_detector = nb.fit(train_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = spam_detector.transform(test_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Bayes:  0.9340036966088454\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "acc_eval = MulticlassClassificationEvaluator(labelCol='label_num')\n",
    "acc = acc_eval.evaluate(results)\n",
    "print ('Bayes: ', acc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Random Forests:  0.8330587931687025\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "spam_detector_rtc = RandomForestClassifier(labelCol='label_num')\n",
    "rtc_model = spam_detector_rtc.fit(train_set)\n",
    "rtc_results = rtc_model.transform(test_set)\n",
    "acc_eval = MulticlassClassificationEvaluator(labelCol='label_num')\n",
    "acc = acc_eval.evaluate(rtc_results)\n",
    "print ('Random Forests: ', acc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Logistic Regression:  0.9722194526491276\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.classification import LogisticRegression\n",
    "spam_detector_lgr = LogisticRegression(labelCol='label_num')\n",
    "lgr_model = spam_detector_lgr.fit(train_set)\n",
    "lgr_results = lgr_model.transform(test_set)\n",
    "acc_eval = MulticlassClassificationEvaluator(labelCol='label_num')\n",
    "acc = acc_eval.evaluate(lgr_results)\n",
    "print ('Logistic Regression: ', acc)"
   ]
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}