I am building an NLP pipeline for Content-Based Email Spam Detection. I am currently running the project on Docker. My issue is that the time taken to train the dataset is taking very long (more than 8 hours). I find this very abnormal as the dataset is only about 33k rows and 3 columns. This happens not only when I use roBERTa embeddings and logistic regression but also in any other combination of embeddings and ML model. Please advise me on why this is occurring. I have provided the code below to reproduce the issue.
Here is my Docker-Compose.yml file
version: "3"
services:
spark:
image: jupyter/all-spark-notebook
container_name: spark-node
depends_on:
- elasticsearch
ports:
- "8888:8888"
- "4040:4040"
volumes:
- .:/home/jovyan/
networks:
- elastic
environment:
- SPARK_WORKER_CORES = 5
- SPARK_WORKER_MEMORY = 11g
- SPARK_DRIVER_MEMORY = 9g
- SPARK_EXECUTOR_MEMORY = 10g
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:7.15.2
container_name: elasticsearch
environment:
- discovery.type=single-node
- ES_JAVA_OPTS=-Xms1g -Xmx1g
- xpack.security.enabled=false
- bootstrap.memory_lock=true
volumes:
- es_data:/usr/share/elasticsearch/data
ports:
- target: 9200
published: 9200
networks:
- elastic
ulimits:
memlock:
soft: -1
hard: -1
kibana:
image: docker.elastic.co/kibana/kibana:7.15.2
container_name: kibana
ports:
- target: 5601
published: 5601
depends_on:
- elasticsearch
networks:
- elastic
environment:
- "ELASTICSEARCH_URL=http://elasticsearch:9200"
- "SERVER_NAME=127.0.0.1"
volumes:
es_data:
driver: local
networks:
elastic:
name: elastic
driver: bridge
This is my Spark Session
spark = SparkSession.builder.master('local[*]') \
.appName('EmailSpamDetection') \
.config('spark.driver.extraClassPath', '/home/jovyan/elasticsearch-spark-30_2.12-7.15.2.jar')\
.config('spark.executor.extraClassPath', '/home/jovyan/elasticsearch-spark-30_2.12-7.15.2.jar')\
.config("spark.es.nodes","elasticsearch")\
.config("spark.executor.memory", "10g") \
.config("spark.driver.memory", "9g") \
.config("spark.worker.memory", "11g") \
.config("spark.cores.max", "16") \
.config("spark.driver.cores", "5") \
.config("spark.executor.cores", "5") \
.config("spark.es.port","9200") \
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
.config("spark.kryoserializer.buffer.max", "2000M")\
.config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:4.1.0") \
.getOrCreate()
Dataset: https://www.kaggle.com/datasets/wanderfj/enron-spam This is the code I used to read the files
def read_files():
email_data_list = []
email_schema = StructType([
StructField('Email', StringType(), True),
StructField('Label', IntegerType(), True),
StructField('Email_ID', IntegerType(), True)])
doc_id=0
for root,dirs,files in os.walk('data'):
if (os.path.split(root)[1] == 'spam'):
for filename in files:
with open(os.path.join(root,filename), "r",encoding="latin-1") as f:
spam_email = f.read()
spam_email = spam_email.replace("\r","").replace("\t","")
spam_email_lines = spam_email.split("\n")
spam_email_final = ""
for line in spam_email_lines:
if line.startswith("Subject:"):
spam_email_final += line.replace("Subject:","").strip()+ " "
if line.startswith("Date:") == False and line.startswith("Cc:") == False and line.startswith("From:") == False and line.startswith("To:") == False and line.startswith("Bcc:") == False and line.startswith("Subject:") == False:
spam_email_final += line + " "
f.close()
if len(spam_email_final) > 20:
doc_id+=1
email_data_list.append((spam_email_final,1,doc_id))
elif (os.path.split(root)[1] == 'ham'):
for filename in files:
with open(os.path.join(root,filename), "r",encoding="latin-1") as f:
ham_email = f.read()
ham_email = ham_email.replace("\r","").replace("\t","")
ham_email_lines = ham_email.split("\n")
ham_email_final = ""
for line in ham_email_lines:
if line.startswith("Subject:"):
ham_email_final += line.replace("Subject:","").strip() + " "
if line.startswith("Date:") == False and line.startswith("Cc:") == False and line.startswith("From:") == False and line.startswith("To:") == False and line.startswith("Bcc:") == False and line.startswith("Subject:") == False:
ham_email_final += line + " "
f.close()
if len(ham_email_final) > 20:
doc_id+=1
email_data_list.append((ham_email_final,0,doc_id))
emailDF = spark.createDataFrame(data=email_data_list, schema = email_schema)
emailDF.show()
return emailDF
emailDF = read_files()
This is the NLP codes
documentAssembler = DocumentAssembler() \
.setInputCol("Email") \
.setOutputCol("documents")
tokenizer = Tokenizer() \
.setInputCols(["documents"]) \
.setOutputCol("tokens")
normalizer = Normalizer() \
.setInputCols(["tokens"]) \
.setOutputCol("tokens_normalized") \
.setLowercase(True) \
.setMinLength(3)
spellchecker = ContextSpellCheckerModel.pretrained() \
.setInputCols(['tokens_normalized'])\
.setOutputCol('tokens_spellchecked')\
.setWordMaxDistance(3) \
from nltk.corpus import stopwords
eng_stopwords = stopwords.words('english')
stopwords_removal = StopWordsCleaner() \
.setInputCols(['tokens_spellchecked']) \
.setOutputCol('tokens_stopwords_removed') \
.setStopWords(eng_stopwords)
lemmatizer = LemmatizerModel.pretrained() \
.setInputCols(['tokens_stopwords_removed']) \
.setOutputCol('tokens_lemmatized')
roberta_word_embeddings = RoBertaEmbeddings.pretrained() \
.setInputCols(["documents",'tokens_lemmatized']) \
.setOutputCol("roberta_embeddings") \
.setCaseSensitive(False)
roberta_embeddings_Sentence = SentenceEmbeddings() \
.setInputCols(["documents", "roberta_embeddings"]) \
.setOutputCol("roberta_sentence_embeddings") \
.setPoolingStrategy("AVERAGE")
roberta_explodeVectors = SQLTransformer(statement=
"SELECT EXPLODE(roberta_sentence_embeddings_finished) AS roberta_embeddings_final, * FROM __THIS__")
lr = LogisticRegression(featuresCol = 'roberta_embeddings_final', labelCol = 'Label', maxIter=10)
roberta_nlp_pipeline = Pipeline().setStages([
documentAssembler,
tokenizer,
normalizer,
spellchecker,
stopwords_removal,
lemmatizer,
roberta_word_embeddings,
roberta_embeddings_Sentence,
roberta_embeddings_finisher,
roberta_explodeVectors,
lr
])
Fitting on Training Data - This part is taking very long due to the logistic regression model included in the pipeline. If I exclude the model, it runs quickly
(trainDF_roberta,testDF_roberta) = emailDF.randomSplit((0.7,0.3),seed=42)
roberta_nlp_model = roberta_nlp_pipeline.fit(trainDF_roberta)