<jupyter_start><jupyter_text>Problem Statement: IMDB Comment Sentiment ClassifierDataset: For this exercise we will use a dataset hosted at http://ai.stanford.edu/~amaas/data/sentiment/**Problem Statement**: This is a dataset for binary sentiment classification containing substantially more data than previous benchmark datasets. We provide a set of 25,000 highly polar movie reviews for training, and 25,000 for testing. There is additional unlabeled data for use as well. Raw text and already processed bag of words formats are provided.  Launch a spark session, verify the spark session UI<jupyter_code>spark.sparkContext.uiWebUrl<jupyter_output><empty_output><jupyter_text>IMDB comments dataset has been stored in the following location<jupyter_code>!wc -l data/imdb-comments.json<jupyter_output>50000 data/imdb-comments.json<jupyter_text>There are 50000 lines in the file. Let's the first line<jupyter_code>!du -sh data/imdb-comments.json<jupyter_output>66M	data/imdb-comments.json<jupyter_text>Total size of the file is 66MB<jupyter_code>!head -n 1 data/imdb-comments.json<jupyter_output>{"label":"test","sentiment":"pos","name":"0_10.txt","content":"I went and saw this movie last night after being coaxed to by a few friends of mine. I'll admit that I was reluctant to see it because from what I knew of <NAME> he was only able to do comedy. I was wrong. Kutcher played the character of <NAME> very well, and <NAME> played <NAME> with such professionalism. The sign of a good movie is that it can toy with our emotions. This one did exactly that. The entire theater (which was sold out) was overcome by laughter during the first half of the movie, and were moved to tears during the second half. While exiting the theater I not only saw many women in tears, but many full grown men as well, trying desperately not to let anyone see them crying. This movie was great, and I suggest that you go see it before you judge."}<jupyter_text>Each line is a self contained json doc. Load the dataset using spark reader specifying the file format as json. As we see above size of the file is 66 MB, we should at least 2 partitons, since I am using dual core system, I will repartition the data to 4. Also will cache the data after repartitioning.<jupyter_code>imdb = spark.read.format("json").load("data/imdb-comments.json").repartition(4).cache()<jupyter_output><empty_output><jupyter_text>Find total number of records<jupyter_code>imdb.count()<jupyter_output><empty_output><jupyter_text>Print Schema and view the field types<jupyter_code>imdb.printSchema()<jupyter_output>root
 |-- content: string (nullable = true)
 |-- label: string (nullable = true)
 |-- name: string (nullable = true)
 |-- sentiment: string (nullable = true)<jupyter_text>Take a look at a few sample data<jupyter_code>imdb.show()<jupyter_output>+--------------------+-----+------------+---------+
|             content|label|        name|sentiment|
+--------------------+-----+------------+---------+
|Actor turned dire...| test| 10000_7.txt|      pos|
|I saw this film o...| test| 10004_9.txt|      pos|
|"Quitting" may be...| test| 10008_8.txt|      pos|
|I must say, every...| test| 10011_9.txt|      pos|
|Even if you're a ...| test| 10015_8.txt|      pos|
|Since this cartoo...| test| 10019_8.txt|      pos|
|This short is one...| test|10022_10.txt|      pos|
|<NAME> t...| test|10026_10.txt|      pos|
|I was fortunate t...| test|  1002_9.txt|      pos|
|The historical in...| test| 10033_8.txt|      pos|
|This movie is goo...| test| 10037_7.txt|      pos|
|Although "They Di...| test|10040_10.txt|      pos|
|<NAME> ...| test| 10044_8.txt|      pos|
|Although this fil...| test| 10048_8.txt|      pos|
|All in all, an ex...| test| 10051_8.txt|      pos|
|i have to rate th...| test|10055_10.txt|      pos|
|Why does ev[...]<jupyter_text>*label* - column indicate whethet the data belong to training or test bucket.*sentiment* - column indicates whether the comment carries positive or negative sentiment. This column has been manually curated.Find out for each combination of label and sentimnet how many records are there.<jupyter_code>from pyspark.sql.functions import *
from pyspark.sql.types import *
imdb.groupBy("sentiment").pivot("label").count().show()<jupyter_output>+---------+-----+-----+
|sentiment| test|train|
+---------+-----+-----+
|      pos|12500|12500|
|      neg|12500|12500|
+---------+-----+-----+<jupyter_text>Look at a sample comment value<jupyter_code>content = imdb.sample(False, 0.001, 1).first().content
content<jupyter_output><empty_output><jupyter_text>Register a UDF function to clean the comment from the html tags. If BeautifulSoup is not installed, you can install it using pip ```(shell command)$ pip install BeautifulSoup4```<jupyter_code>from bs4 import BeautifulSoup
from pyspark.sql.types import * 
import re
def remove_html_tags(text):
    text = BeautifulSoup(text, "html5lib").text.lower() #removed html tags
    text = re.sub("[\W]+", " ", text)
    return text

spark.udf.register("remove_html_tags", remove_html_tags, StringType())<jupyter_output><empty_output><jupyter_text>Test the remove_html_tags function<jupyter_code>remove_html_tags(content)<jupyter_output><empty_output><jupyter_text>Apply the the udf on the imdb dataframe.<jupyter_code>imdb_clean = imdb.withColumn("content", expr("remove_html_tags(content)")).cache()
imdb_clean.sample(False, 0.001, 1).first().content<jupyter_output><empty_output><jupyter_text>Use Tokenizer to split the string into terms. Then use StopWordsRemover to remove stop words like prepositions, apply CountVectorizer to find all distinct terms and found of each term per document.<jupyter_code>from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer, StopWordsRemover
tokenizer = Tokenizer(inputCol="content", outputCol="terms")
terms_data = tokenizer.transform(imdb_clean)
print(terms_data.sample(False, 0.001, 1).first().terms)
remover = StopWordsRemover(inputCol="terms", outputCol="filtered")
terms_stop_removed = remover.transform(terms_data)
print(terms_stop_removed.sample(False, 0.001, 1).first().filtered)
count_vectorizer = CountVectorizer(inputCol="filtered", outputCol="count_vectors")
count_vectorizer_model = count_vectorizer.fit(terms_stop_removed)
count_vectorized = count_vectorizer_model.transform(terms_stop_removed)
count_vectorized.sample(False, 0.001, 1).first().count_vectors<jupyter_output><empty_output><jupyter_text>count_vectorized Dataframe contains a column count_vectors that is a SparseVector representing which term appears and how many times. The key is the index of all unique terms. You can find list of terms count_vectorizer_model.vocabulary. See below.<jupyter_code>print(count_vectorizer_model.vocabulary[:100], "\n\nTotal no of terms", len(count_vectorizer_model.vocabulary))
count_vectorized.show()<jupyter_output>+--------------------+-----+------------+---------+--------------------+--------------------+--------------------+
|             content|label|        name|sentiment|               terms|            filtered|       count_vectors|
+--------------------+-----+------------+---------+--------------------+--------------------+--------------------+
|actor turned dire...| test| 10000_7.txt|      pos|[actor, turned, d...|[actor, turned, d...|(103999,[1,4,7,10...|
|i saw this film o...| test| 10004_9.txt|      pos|[i, saw, this, fi...|[saw, film, septe...|(103999,[0,1,2,5,...|
| quitting may be ...| test| 10008_8.txt|      pos|[, quitting, may,...|[, quitting, may,...|(103999,[0,2,3,4,...|
|i must say every ...| test| 10011_9.txt|      pos|[i, must, say, ev...|[must, say, every...|(103999,[0,2,5,7,...|
|even if you re a ...| test| 10015_8.txt|      pos|[even, if, you, r...|[even, re, fan, j...|(103999,[0,1,2,6,...|
|since this cartoo...| test| 10019_8.txt|      pos|[since, this, car...|[since, [...]<jupyter_text>SparkVector represents a vector of 103999, that means in the dataset (corpus) there are 103999 unique terms. Per document, only a few will be present. Find density of each count_vectors.<jupyter_code>vocab_len = len(count_vectorizer_model.vocabulary)
spark.udf.register("density", lambda r: r.numNonzeros() / vocab_len, DoubleType())
count_vectorized.select(expr("density(count_vectors) density")).show()<jupyter_output>+--------------------+
|             density|
+--------------------+
|0.001615400148078...|
|0.001375013221280...|
|0.003884652737045...|
|6.730833950326445E-4|
|0.003326955066875643|
|8.269310281829633E-4|
|  6.0577505552938E-4|
|0.003951961076548813|
|5.769286243136953E-4|
|5.865441013855903E-4|
|0.001163472725699...|
|3.365416975163222...|
|0.002115404955816...|
|5.096202848104309E-4|
|0.002144251387032...|
|7.019298262483293E-4|
|3.557726516601121E-4|
|6.538524408888547E-4|
|6.826988721045394E-4|
|0.001923095414378...|
+--------------------+
only showing top 20 rows<jupyter_text>Density report shows, the count_vectors has very low density which illustrate the benefit of the choice of DenseVector for this column. Now, calculate tfidf for the document.<jupyter_code>idf = IDF(inputCol="count_vectors", outputCol="features")
idf_model = idf.fit(count_vectorized)
idf_data = idf_model.transform(count_vectorized)
idf_data.sample(False, 0.001, 1).first().features
idf_data.printSchema()<jupyter_output>root
 |-- content: string (nullable = true)
 |-- label: string (nullable = true)
 |-- name: string (nullable = true)
 |-- sentiment: string (nullable = true)
 |-- terms: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- filtered: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- count_vectors: vector (nullable = true)
 |-- features: vector (nullable = true)<jupyter_text>Apply StringIndexer to conver the sentiment column from String type to number type - this is prerequisit to apply the LogisticRegression algorithm.<jupyter_code>from pyspark.ml.feature import StringIndexer
string_indexer = StringIndexer(inputCol="sentiment", outputCol="sentiment_idx")
string_indexer_model = string_indexer.fit(idf_data)
label_encoded = string_indexer_model.transform(idf_data)
label_encoded.select("sentiment", "sentiment_idx").show()<jupyter_output>+---------+-------------+
|sentiment|sentiment_idx|
+---------+-------------+
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
|      pos|          0.0|
+---------+-------------+
only showing top 20 rows<jupyter_text>Split the data into traininf and testing groups with 70/30 ratio. Cache the dataframe so that training runs faster.<jupyter_code>training, testing = label_encoded.randomSplit(weights=[0.7, 0.3], seed=1)
training.cache()
testing.cache()<jupyter_output><empty_output><jupyter_text>Verify that the StringIndex has done the expected job and training and testing data maintain the ratio of positive and negative records as in the whole dataset.<jupyter_code>training.groupBy("sentiment_idx", "sentiment").count().show()
testing.groupBy("sentiment_idx", "sentiment").count().show()<jupyter_output>+-------------+---------+-----+
|sentiment_idx|sentiment|count|
+-------------+---------+-----+
|          1.0|      neg| 7412|
|          0.0|      pos| 7507|
+-------------+---------+-----+<jupyter_text>Apply LogisticRegression classifier<jupyter_code>from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(maxIter=10000, regParam=0.1, elasticNetParam=0.0, 
                        featuresCol="features", labelCol="sentiment_idx")<jupyter_output><empty_output><jupyter_text>Show the parameters that the LogisticRegression classifier takes.<jupyter_code>print(lr.explainParams())
lr_model = lr.fit(training)
lr_model.coefficients[:100]<jupyter_output><empty_output><jupyter_text>From the training summary find out the cost decay of the model.<jupyter_code>training_summary = lr_model.summary
import pandas as pd
import matplotlib.pyplot as plt

%matplotlib inline
pd.Series(training_summary.objectiveHistory).plot()
plt.xlabel("Iteration")
plt.ylabel("Cost")<jupyter_output><empty_output><jupyter_text>Find area under the curve. Closer to 1 is better<jupyter_code>training_summary.areaUnderROC
predictions = lr_model.transform(testing).withColumn("match", expr("prediction == sentiment_idx"))
predictions.select("prediction", "sentiment_idx", "sentiment", "match").sample(False, 0.01).show(10)
predictions.groupBy("sentiment_idx").pivot("prediction").count().show()<jupyter_output>+-------------+----+----+
|sentiment_idx| 0.0| 1.0|
+-------------+----+----+
|          0.0|6779| 728|
|          1.0| 890|6522|
+-------------+----+----+<jupyter_text>Find the accuracy of the prediction<jupyter_code>accuracy = predictions.select(expr("sum(cast(match as int))")).first()[0] / predictions.count()
accuracy<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16989996968778417, 'fraction_numerical': 0.05509245225826008, 'mean_word_length': 3.5522593997930323, 'pattern_counts': {'":': 4, '<': 95, '<?xml version=': 0, '>': 95, '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': '25', 'dir': 'jupyter-structured-clean-dedup', 'id': '519253', 'license': 'Apache-2.0', 'n_tokens_mistral': 4696, 'n_tokens_neox': 4243, 'n_words': 1121, 'path': 'SparkML - 04 Text_Analysis.ipynb', 'repo_name': 'nitturis/pyspark-examples', 'size': '13241'}