Spam Classification Using PySpark in Python
By Soham DasIn this tutorial, we will build a spam classifier in Python using Apache Spark which can tell whether a given message is spam or not! I have used a classic spam-ham dataset from the UCI dataset repository for this tutorial.
Spam Classifier Using PySpark
So we will first create a spark session and import the data and then rename the columns for ease of use.
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('nlp').getOrCreate()
data = spark.read.csv("smsspamcollection/SMSSpamCollection",inferSchema=True,sep='\t')
data = data.withColumnRenamed('_c0','class').withColumnRenamed('_c1','text')Let’s just have a look at our data.
data.show()
+-----+--------------------+ |class| text| +-----+--------------------+ | ham|Go until jurong p...| | ham|Ok lar... Joking ...| | spam|Free entry in 2 a...| | ham|U dun say so earl...| | ham|Nah I don't think...| | spam|FreeMsg Hey there...| | ham|Even my brother i...| | ham|As per your reque...| | spam|WINNER!! As a val...| | spam|Had your mobile 1...| | ham|I'm gonna be home...| | spam|SIX chances to wi...| | spam|URGENT! You have ...| | ham|I've been searchi...| | ham|I HAVE A DATE ON ...| | spam|XXXMobileMovieClu...| | ham|Oh k...i'm watchi...| | ham|Eh u remember how...| | ham|Fine if thats th...| | spam|England v Macedon...| +-----+--------------------+ only showing top 20 rows
Clean and Prepare the Data
Creating a new length feature:
from pyspark.sql.functions import length
data = data.withColumn('length',length(data['text']))
data.show()+-----+--------------------+------+ |class| text|length| +-----+--------------------+------+ | ham|Go until jurong p...| 111| | ham|Ok lar... Joking ...| 29| | spam|Free entry in 2 a...| 155| | ham|U dun say so earl...| 49| | ham|Nah I don't think...| 61| | spam|FreeMsg Hey there...| 147| | ham|Even my brother i...| 77| | ham|As per your reque...| 160| | spam|WINNER!! As a val...| 157| | spam|Had your mobile 1...| 154| | ham|I'm gonna be home...| 109| | spam|SIX chances to wi...| 136| | spam|URGENT! You have ...| 155| | ham|I've been searchi...| 196| | ham|I HAVE A DATE ON ...| 35| | spam|XXXMobileMovieClu...| 149| | ham|Oh k...i'm watchi...| 26| | ham|Eh u remember how...| 81| | ham|Fine if thats th...| 56| | spam|England v Macedon...| 155| +-----+--------------------+------+ only showing top 20 rows
If you observe carefully the difference is pretty clear and obvious. Let’s use the group-by method use to confirm this trend:
data.groupby('class').mean().show()
+-----+-----------------+ |class| avg(length)| +-----+-----------------+ | ham|71.45431945307645| | spam|138.6706827309237| +-----+-----------------+
An average difference of 139 and 71 is big enough to use it as a feature in modeling.
Feature Transformations
Now we are going to tokenize the words, remove stopwords, and implement TF-IDF step by step and then convert them into a single dense vector using the vector-assembler.
from pyspark.ml.feature import Tokenizer,StopWordsRemover, CountVectorizer,IDF,StringIndexer from pyspark.ml.feature import VectorAssembler from pyspark.ml.linalg import Vector tokenizer = Tokenizer(inputCol="text", outputCol="token_text") stopremove = StopWordsRemover(inputCol='token_text',outputCol='stop_tokens') count_vec = CountVectorizer(inputCol='stop_tokens',outputCol='c_vec') idf = IDF(inputCol="c_vec", outputCol="tf_idf") ham_spam_to_num = StringIndexer(inputCol='class',outputCol='label') clean_up = VectorAssembler(inputCols=['tf_idf','length'],outputCol='features')
Model and Pipeline
We are going to use the Naive-Bayes classifier as our model for this tutorial.
from pyspark.ml.classification import NaiveBayes # Use defaults nb = NaiveBayes()
Pipelining is important so that we don’t have to re-iterate the previous preprocessing steps for further experimentation.
from pyspark.ml import Pipeline data_prep_pipe = Pipeline(stages=[ham_spam_to_num,tokenizer,stopremove,count_vec,idf,clean_up]) cleaner = data_prep_pipe.fit(data) clean_data = cleaner.transform(data)
Training and Evaluation
Let’s just quickly check that all the preprocessing was done correctly.
clean_data = clean_data.select(['label','features']) clean_data.show()
+-----+--------------------+ |label| features| +-----+--------------------+ | 0.0|(13424,[7,11,31,6...| | 0.0|(13424,[0,24,297,...| | 1.0|(13424,[2,13,19,3...| | 0.0|(13424,[0,70,80,1...| | 0.0|(13424,[36,134,31...| | 1.0|(13424,[10,60,139...| | 0.0|(13424,[10,53,103...| | 0.0|(13424,[125,184,4...| | 1.0|(13424,[1,47,118,...| | 1.0|(13424,[0,1,13,27...| | 0.0|(13424,[18,43,120...| | 1.0|(13424,[8,17,37,8...| | 1.0|(13424,[13,30,47,...| | 0.0|(13424,[39,96,217...| | 0.0|(13424,[552,1697,...| | 1.0|(13424,[30,109,11...| | 0.0|(13424,[82,214,47...| | 0.0|(13424,[0,2,49,13...| | 0.0|(13424,[0,74,105,...| | 1.0|(13424,[4,30,33,5...| +-----+--------------------+ only showing top 20 rows
Now we are all set for training and evaluation with a 70:30 split.
(training,testing) = clean_data.randomSplit([0.7,0.3]) spam_predictor = nb.fit(training) test_results = spam_predictor.transform(testing) test_results.show()
+-----+--------------------+--------------------+--------------------+----------+ |label| features| rawPrediction| probability|prediction| +-----+--------------------+--------------------+--------------------+----------+ | 0.0|(13424,[0,1,3,9,1...|[-572.06653080640...|[1.0,3.2853946379...| 0.0| | 0.0|(13424,[0,1,5,15,...|[-1001.2595934260...|[1.0,2.7574544759...| 0.0| | 0.0|(13424,[0,1,5,20,...|[-806.54241219940...|[1.0,1.8852085844...| 0.0| | 0.0|(13424,[0,1,9,14,...|[-537.01474869015...|[1.0,2.8233277769...| 0.0| | 0.0|(13424,[0,1,11,32...|[-869.75284680397...|[1.0,1.0200416791...| 0.0| | 0.0|(13424,[0,1,14,31...|[-215.73138546316...|[1.0,5.8314497532...| 0.0| | 0.0|(13424,[0,1,14,78...|[-686.70388741456...|[1.0,2.0779668967...| 0.0| | 0.0|(13424,[0,1,23,63...|[-1310.9094107562...|[1.0,4.4866463813...| 0.0| | 0.0|(13424,[0,1,24,31...|[-356.01275797052...|[1.0,3.5064139479...| 0.0| | 0.0|(13424,[0,1,27,35...|[-1493.1508968151...|[0.99999997768200...| 0.0| | 0.0|(13424,[0,1,30,12...|[-599.21107138763...|[1.0,2.9152869776...| 0.0| | 0.0|(13424,[0,1,46,17...|[-1139.5052371653...|[9.30530122501920...| 1.0| | 0.0|(13424,[0,1,146,1...|[-254.82374909461...|[0.20196018944218...| 1.0| | 0.0|(13424,[0,2,3,6,9...|[-3301.7237938480...|[1.0,1.0067276963...| 0.0| | 0.0|(13424,[0,2,4,7,2...|[-511.87873676486...|[1.0,1.3675876660...| 0.0| | 0.0|(13424,[0,2,4,8,1...|[-1316.0759246967...|[1.0,1.3703321229...| 0.0| | 0.0|(13424,[0,2,4,40,...|[-1582.4686915061...|[0.99999999442732...| 0.0| | 0.0|(13424,[0,2,4,44,...|[-1909.3003347074...|[1.0,4.9564372811...| 0.0| | 0.0|(13424,[0,2,7,11,...|[-851.30100707005...|[1.0,3.9992581112...| 0.0| | 0.0|(13424,[0,2,7,43,...|[-590.79005982680...|[1.0,5.8169836805...| 0.0| +-----+--------------------+--------------------+--------------------+----------+ only showing top 20 rows
That was not very intuitive though! Let’s use the MulticlassClassificationEvaluator function to extract more meaning out of the results.
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
acc_eval = MulticlassClassificationEvaluator()
acc = acc_eval.evaluate(test_results)
print("Accuracy of model at predicting spam was: {}".format(acc))
Accuracy of model at predicting spam was: 0.9210916371646145
92% accuracy is not very bad. Hopefully, this tutorial helped you to start working on some NLP stuff using Spark. This dataset is obviously not ideal for using Spark as it is a Big Data framework but it still serves for demonstration purposes. You can get the actual notebook for the code here.
Leave a Reply