<jupyter_start><jupyter_text>Environment Setup<jupyter_code>#!pip install spark-nlp
#!pip install fastparquet 
#!pip install spark-nlp==2.6.1
import re
import numpy as np
import pandas as pd

from pyspark import SparkContext
from pyspark.sql import SparkSession, SQLContext
from pyspark.ml import Pipeline

import sparknlp
from sparknlp.base import *
from sparknlp.annotator import *

from pyspark.sql.functions import udf
import pyspark.sql.types as T

import pyspark.sql.functions as F
from pyspark.sql.types import *

from pyspark.ml.feature import HashingTF, IDF, StringIndexer, SQLTransformer,IndexToString
from pyspark.ml.feature import CountVectorizer , IDF
from pyspark.ml.clustering import LDA

from utils import CUSTOM_STOP_WORDS
spark = SparkSession.builder \
    .appName("Spark NLP")\
    .master("local[4]")\
    .config("spark.driver.memory","16G")\
    .config("spark.driver.maxResultSize", "0") \
    .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.11:2.6.1")\
    .config("spark.kryoserializer.buffer.max", "1000M")\
    .getOrCreate()

spark
"""
from pyspark.sql import SparkSession

# start spark session configured for spark nlp
spark = SparkSession.builder \
     .master('local[*]') \
     .appName('Spark NLP') \
     .config('spark.jars.packages') \
     .getOrCreate()

spark
"""
print()
#spark = sparknlp.start()

#spark
sc = spark
sqlContext = SQLContext(spark)
sqlContext
#sc = SparkContext('local', 'PySPARK LDA Example')
#sqlContext = SQLContext(sc)<jupyter_output><empty_output><jupyter_text>Load data<jupyter_code>df_spark = sqlContext.read.parquet("newsgroup_20_data.parquet")
print(df_spark.count())
#df_spark.head(2)
d = df_spark.head(1)
d = d[0]

print(d.asDict()['category'])
#print(d.asDict()['news'])
#(trainingData, testData) = df_spark.randomSplit([0.7, 0.3], seed = 100)
trainingData = df_spark.copy()<jupyter_output><empty_output><jupyter_text>Pre-Process Data  1. data cleaning<jupyter_code>col_input = "news"
col_label = "category"
col_nlp = 'col_nlp'
def text_cleaner(sentence):
    
    # clean the punctuations
    punc_re = r'[^a-zA-Z0-9 &]'
    sentence = re.sub(punc_re, ' ', sentence)
    
    # tokens
    arr = sentence.split()
    
    # remove white spaces
    # lowercase
    # filter words having lenght <= 3
    arr = [word.strip().lower() for word in arr if word.isalpha() and len(word)>=4]
    
    arr = " ".join(arr)
    return arr
#data = trainingData.limit(1000)
data = trainingData

udf_text_cleaner = F.udf(text_cleaner, StringType())

data_train_clean = data.withColumn(col_nlp, udf_text_cleaner(col_input))

print(data_train_clean.count())
data_train_clean.limit(2).show()<jupyter_output>1000
+--------------------+------------+--------------------+
|                news|    category|             col_nlp|
+--------------------+------------+--------------------+
| agate!ames!purdu...|misc.forsale|agate ames purdue...|
| agate!iat.holone...|   rec.autos|agate holonet psi...|
+--------------------+------------+--------------------+<jupyter_text>2. vectorizer<jupyter_code>def type_changer(sentence):
    return sentence.split(" ")
udf_type_changer = F.udf(type_changer, ArrayType(elementType=StringType()))

# get tokens
data_arr = data_train_clean.withColumn("col_nlp_arr", udf_type_changer(col_nlp))

# TF
cv = CountVectorizer(inputCol="col_nlp_arr", outputCol="raw_features", vocabSize=5000, minDF=10.0)
cvmodel = cv.fit(data_arr)
result_cv = cvmodel.transform(data_arr)

# IDF
idf = IDF(inputCol="raw_features", outputCol="features")
idfModel = idf.fit(result_cv)
result_tfidf = idfModel.transform(result_cv)

pp_train_data = result_tfidf.select('category', col_input, "features")

print(type(pp_train_data))
pp_train_data.limit(2).show()<jupyter_output><class 'pyspark.sql.dataframe.DataFrame'>
+------------+--------------------+--------------------+
|    category|                news|            features|
+------------+--------------------+--------------------+
|misc.forsale| agate!ames!purdu...|(2341,[1,5,9,10,1...|
|   rec.autos| agate!iat.holone...|(2341,[2,4,5,8,9,...|
+------------+--------------------+--------------------+<jupyter_text>LDA Model  1. train model<jupyter_code>numTopics = 20 # number of topics
 
lda = LDA(k=numTopics, seed = 1, optimizer="online", optimizeDocConcentration=True,
          maxIter = 10,           # number of iterations
          learningDecay = 0.51,   # kappa, learning rate
          learningOffset = 64.0,  # tau_0, larger values downweigh early iterations
          subsamplingRate = 0.05, # mini batch fraction 
          )
 
model = lda.fit(pp_train_data)
 
ll = model.logLikelihood(pp_train_data)
lp = model.logPerplexity(pp_train_data)
print("The lower bound on the log likelihood of the entire corpus: " + str(ll))
print("The upper bound on perplexity: " + str(lp))<jupyter_output>The lower bound on the log likelihood of the entire corpus: -2617672.0557389343
The upper bound on perplexity: 7.414561064286616<jupyter_text>2. topic insights<jupyter_code>model.vocabSize()
model.describeTopics().first()
print("The topics described by their top-weighted terms:")
model.describeTopics(5).limit(6).show()
model.topicsMatrix()<jupyter_output><empty_output><jupyter_text>3. topic assignment<jupyter_code>max_index = F.udf(lambda x: x.tolist().index(max(x)), IntegerType())


lda_train_data = model.transform(pp_train_data)
lda_train_data = lda_train_data.withColumn("topicID", max_index("topicDistribution"))

print(lda_train_data.count())
lda_train_data.limit(2).show()
"""
topicDistribution : list of topic weights (len==num_topics)
"""
print()<jupyter_output><jupyter_text>4. topic model assesment<jupyter_code>#lda_train_data.first()
X_topics = lda_train_data.select("category", "topicID").toPandas()

print(X_topics.shape)
X_topics.head(2)
def topic_metrics(df):
    #print(df.head(2))
    arr = df["topicID"].value_counts()
    max_topic = arr.index.values[0]
    perc_dominance = arr[max_topic] / arr.sum()
    
    result = pd.Series(data=[int(max_topic), perc_dominance], index=["category_pred", "perc_dominance"])
    
    return result
X = X_topics.copy()
#X = X_topics.head(10)

X_label_mapping = X.groupby("category").apply(topic_metrics).reset_index()
X_label_mapping["category_pred"] = X_label_mapping["category_pred"].astype("int")
X_label_mapping["perc_dominance"] = np.round(X_label_mapping["perc_dominance"], 2)
X_label_mapping
dict_mapper = {}
for i in zip(X_label_mapping["category"], X_label_mapping["category_pred"]):
    dict_mapper[i[0]] = int(i[1])
    
X = X_topics.copy()
X["category_pred"] = X["topicID"].replace(dict_mapper)

print(X.shape)
X.head(2)
X_topics.info()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18255395683453238, 'fraction_numerical': 0.020233812949640287, 'mean_word_length': 3.928360413589365, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '166381', 'license': 'MIT', 'n_tokens_mistral': 2478, 'n_tokens_neox': 2341, 'n_words': 418, 'path': 'spark/dep_nlp_clustering_20NewsGroup_v4.ipynb', 'repo_name': 'VijaySingh-GSLab/aws-ml', 'size': '6672'}