# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Introduction to Spark ML: An application to Sentiment Analysis

# # Spark ML

# In previous versions of Spark, most Machine Learning funcionality was provided through RDD (Resilient Distributed Datasets). However, to improve performance and communicability of results, Spark developers ported the ML functionality to work almost exclusively with DataFrames. Future releases of Spark will not update the support of ML with RDDs.
#
# In this modern Spark ML approach, there are _Estimators_ and _Transformers_. Estimators have some parameters that need to be fit into the data. After fitting, Estimators return Transformers. Tranformers can be applied to dataframes, taking one (or several) columns as input and creating (or several) columns as output.
#
# A _Pipeline_ combines several _Tranformers_ with a final _Estimator_. The _Pipeline_, therefore, can be fit to the data because the final step of the process (the _Estimator_) is fit to the data. The result of the fitting is a pipelined _Transformer_ that takes an input dataframe through all the stages of the Pipeline.
#
# There is a third type of functionality that allows to select features.

# For example, for analyzing text, a typical pipelined estimator is as follows:
#
# <img src="http://spark.apache.org/docs/latest/img/ml-Pipeline.png" alt="ML Pipeline" style="width: 100%;"/>

# After fitting, the Pipeline becomes a transformer:
#
# <img src="http://spark.apache.org/docs/latest/img/ml-PipelineModel.png" alt="ML Model" style="width: 100%;"/>
# (Images from http://spark.apache.org/docs/latest/ml-pipeline.html)

# Importantly, transformers can be saved and exchanged with other data scientists, improving reproducibility.

# ## Loading packages and connecting to Spark cluster

from pyspark.sql import SparkSession
from pyspark.ml import feature, regression, evaluation, Pipeline
from pyspark.sql import functions as fn, Row
import matplotlib.pyplot as plt
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext

import numpy as np

# dataframe functions
from pyspark.sql import functions as fn
from __future__ import division

# ### Transformers and Estimators

# Let's define an example dataframe with documents

documents_df = spark.sparkContext.parallelize([
        [1, 'cats are cute', 0],
        [2, 'dogs are playfull', 0],
        [3, 'lions are big', 1],
        [4, 'cars are fast', 1]]).toDF(['doc_id', 'text', 'user_id'])

# There are several ways of transforming the data from raw input to something that can be analyzed with a statistical model.
#
# Some examples of such transformers are displayed below:
#
# #### Tokenizer
#
# Suppose that we want to split the words or _tokens_ of a document. This is what `Tokenizer` does.

from pyspark.ml.feature import Tokenizer

# Almost all transfomers and estimator require you to specificy the input column of the dataframe and the output column that will be added to the dataframe.

# the tokenizer object
tokenizer = Tokenizer().setInputCol('text').setOutputCol('words')

# We can now transform the dataframe

tokenizer.transform(documents_df).show()

# #### CountVectorizer

# This transformer counts how many times a word appears in a list and produces a vector with such counts. This is very useful for text analysis.

from pyspark.ml.feature import CountVectorizer

# A `CountVectorizer` is different from a `Tokenizer` because it needs to learn how many different tokens there are in the input column. With that number, it will output vectors with consistent dimensions. Therefore, `CountVectorizer` is an `Estimator` that, when fitted, returns a `Transformer`.

count_vectorizer_estimator = CountVectorizer().setInputCol('words').setOutputCol('features')

# Now we need to user the words column that  generated by the `tokenizer` transformer

count_vectorizer_transformer = count_vectorizer_estimator.fit(tokenizer.transform(documents_df))

# which results in:

count_vectorizer_transformer.transform(tokenizer.transform(documents_df)).show(truncate=False)

# The column `features` is a sparse vector representation. For example, for the first document, we have three features present: 0, 3, and 5. By looking at the vocabulary learned by `count_vectorizer_transformer`, we can know which words those feature indices refer to:

# list of words in the vocabulary
count_vectorizer_transformer.vocabulary

# ## Pipelines
#
# Sometimes, we have long preprocessing steps that take raw data and transform it through several stages. As explained before, these complex transformations can be captured by Pipelines.
#
# Pipelines are always estimators, even when they contain several transformers. After a pipeline is `fit` to the data, the pipeline becomes an transformer.
#
# We will now define a pipeline that takes the raw `text` column and produces the `features` column previously explained

from pyspark.ml import Pipeline

pipeline_cv_estimator = Pipeline(stages=[tokenizer, count_vectorizer_estimator])

pipeline_cv_transformer = pipeline_cv_estimator.fit(documents_df)

pipeline_cv_transformer.transform(documents_df).show()

# In more complex scenarios, you can even chain Pipeline transformers. We will see this case in the actual use case below.
#
# For a more detail explanation of Pipelines, Estimators, and Transformers, [see here](http://spark.apache.org/docs/latest/ml-pipeline.html)

# ## Load sentiment data

import pandas as pd

sentiments_df = spark.read.parquet('datasets/sentiments.parquet')

sentiments_df.printSchema()

# The schema is very simple: for each word, we have whether it is positive (+1) or negative (-1)

# a sample of positive words
sentiments_df.where(fn.col('sentiment') == 1).show(5)

# a sample of negative words
sentiments_df.where(fn.col('sentiment') == -1).show(5)

# Lets see how many of each category we have

sentiments_df.groupBy('sentiment').agg(fn.count('*')).show()

# We have almost two times the number of negative words!

# # A simple approach to sentiment analysis

# One simple approach for sentiment analysis is to simple count the number of positive and negative words in a text and then compute the average sentiment. Assuming that positive words are +1 and negative words are -1, we can classify a text as positive if the average sentiment is greater than zero and negative otherwise

# To test our approach, we will use a sample of [IMDB](http://www.imdb.com/) reviews that were tagged as positive and negative.
#
# Let's load them:

imdb_reviews_df = spark.read.parquet('datasets/imdb_reviews_preprocessed.parquet')

# Let's take a look at a positive review

imdb_reviews_df.where(fn.col('score') == 1).first()

# And a negative one

imdb_reviews_df.where(fn.col('score') == 0).first()

# The first problem that we encounter is that the reviews are in plain text. We need to split the words and then match them to `sentiment_df`.

# To do, we will use a transformation that takes raw text and outputs a list of words

from pyspark.ml.feature import RegexTokenizer

# `RegexTokenizer` extracts a sequence of matches from the input text. Regular expressions are a powerful tool to extract strings with certain characteristics.

tokenizer = RegexTokenizer().setGaps(False)\
  .setPattern("\\p{L}+")\
  .setInputCol("review")\
  .setOutputCol("words")

# The pattern `\p{L}+` means that it will extract letters without accents (e.g., it will extract "Acuna" from "Acuña"). `setGaps` means that it will keep applying the rule until it can't extract new words. You have to set the input column from the incoming dataframe (in our case the `review` column) and the new column that will be added (e.g., `words`).
#
# We are ready to transform the input dataframe `imdb_reviews_df` with the tokenizer:

review_words_df = tokenizer.transform(imdb_reviews_df)
print(review_words_df)

# Applying the transformation doesn't actually do anything until you apply an action. But as you can see, a new column `words` of type `array` of `string` was added by the transformation. We can see how it looks:

review_words_df.show(5)

# Now, we want to match every word from `sentiment_df` in the array `words` shown before. One way of doing this is to _explode_ the column `words` to create a row for each element in that list. Then, we would join that result with the dataframe `sentiment` to continue further.

review_words_df.select('id', fn.explode('words').alias('word')).show(5)

# Now if we join that with sentiment, we can see if there are positive and negative words in each review:

review_word_sentiment_df = review_words_df.\
    select('id', fn.explode('words').alias('word')).\
    join(sentiments_df, 'word')
review_word_sentiment_df.show(5)

# Now we can simply average the sentiment per review id and, say, pick positive when the average is above 0, and negative otherwise.

simple_sentiment_prediction_df = review_word_sentiment_df.\
    groupBy('id').\
    agg(fn.avg('sentiment').alias('avg_sentiment')).\
    withColumn('predicted', fn.when(fn.col('avg_sentiment') > 0, 1.0).otherwise(0.))
simple_sentiment_prediction_df.show(5)

# Now, lets compute the accuracy of our prediction

imdb_reviews_df.\
    join(simple_sentiment_prediction_df, 'id').\
    select(fn.expr('float(score = predicted)').alias('correct')).\
    select(fn.avg('correct')).\
    show()

# Not bad with such a simple approach! But can we do better than this?

# ## A data-driven sentiment prediction

# There are couple of problems with the previous approach:
# 1. Positive and negative words had the same weight (e.g., good == amazing)
# 1. Maybe a couple of negative words make the entire review negative, whereas positive words do not
# 1. While our dataset is artificially balanced (12500 positive and 12500 negative), there are usually more positive than negative reviews, and therefore we should bias our predictions towards positive ones.

# We could use __data__ to estimate the sentiment that each word is contributing to the final sentiment of a review. Given that we are trying to predict negative and positve reviews, then we can use logistic regression for such binary prediction.

# ### From text to numerical features

# One typical approach is to count how many times a word appears in the text and then perform a reweighting so that words that are very common are "counted" less.

# In Spark, we can achieve this by using several transformers:
#
# __Raw text => Tokens => Remove stop words => Term Frequency => Reweighting by Inverse Document frequency__

# To perform this sequence we will create a __`Pipeline`__ to consistently represent the steps from raw text to TF-IDF.

# First, we need to create a sequence to take from raw text to term frequency. This is necessary because we don't know the number of tokens in the text and therefore we need to _estimate_ such quantity from the data.

# we obtain the stop words from a website
import requests
stop_words = requests.get('http://ir.dcs.gla.ac.uk/resources/linguistic_utils/stop_words').text.split()
stop_words[0:10]

from pyspark.ml.feature import StopWordsRemover
sw_filter = StopWordsRemover()\
  .setStopWords(stop_words)\
  .setCaseSensitive(False)\
  .setInputCol("words")\
  .setOutputCol("filtered")

# Finally, for this initial `Pipeline`, we define a counter vectorizer estimator

# +
from pyspark.ml.feature import CountVectorizer

# we will remove words that appear in 5 docs or less
cv = CountVectorizer(minTF=1., minDF=5., vocabSize=2**17)\
  .setInputCol("filtered")\
  .setOutputCol("tf")
# -

# we now create a pipelined transformer
cv_pipeline = Pipeline(stages=[tokenizer, sw_filter, cv]).fit(imdb_reviews_df)

# now we can make the transformation between the raw text and the counts
cv_pipeline.transform(imdb_reviews_df).show(5)

# The term frequency vector is represented with a sparse vector. We have 26,677 terms.

len(cv_pipeline.stages[-1].vocabulary)

# Finally, we build another pipeline that takes the output of the previous pipeline and _lowers_ the terms of documents that are very common.

from pyspark.ml.feature import IDF
idf = IDF().\
    setInputCol('tf').\
    setOutputCol('tfidf')

idf_pipeline = Pipeline(stages=[cv_pipeline, idf]).fit(imdb_reviews_df)

idf_pipeline.transform(imdb_reviews_df).show(5)

# Therefore, the `idf_pipeline` takes the raw text from the datafarme `imdb_reviews_df` and creates a feature vector vector called `tfidf`!

tfidf_df = idf_pipeline.transform(imdb_reviews_df)

# # Data science pipeline for estimating sentiments

# First, let's split the data into training, validation, and testing.

training_df, validation_df, testing_df = imdb_reviews_df.randomSplit([0.6, 0.3, 0.1], seed=0)

[training_df.count(), validation_df.count(), testing_df.count()]

# One immediately apparent problem is that the number of features in the dataset is far larger than the number of training examples. This can lead to serious overfitting.
#
# Let's look at this more closely. Let's apply a simple prediction model known as logistic regression.
#
# [Logistic regression](https://en.wikipedia.org/wiki/Logistic_regression) will take the `tfidf` features and predict whether the review is positive (`score == 1`) or negative (`score == 0`).

from pyspark.ml.classification import LogisticRegression

lr = LogisticRegression().\
    setLabelCol('score').\
    setFeaturesCol('tfidf').\
    setRegParam(0.0).\
    setMaxIter(100).\
    setElasticNetParam(0.)

# Lets create a pipeline transformation by chaining the `idf_pipeline` with the logistic regression step (`lr`)

lr_pipeline = Pipeline(stages=[idf_pipeline, lr]).fit(training_df)

# Lets estimate the accuracy:

lr_pipeline.transform(validation_df).\
    select(fn.expr('float(prediction = score)').alias('correct')).\
    select(fn.avg('correct')).show()

# The performance is much better than before.
#
# The problem however is that we are overfitting because we have many features compared to the training examples:

# For example, if we look at the weights of the features, there is a lot of noise:

import pandas as pd
vocabulary = idf_pipeline.stages[0].stages[-1].vocabulary
weights = lr_pipeline.stages[-1].coefficients.toArray()
coeffs_df = pd.DataFrame({'word': vocabulary, 'weight': weights})

# The most negative words are:

coeffs_df.sort_values('weight').head(5)

# And the most positive:

coeffs_df.sort_values('weight', ascending=False).head(5)

# But none of them make sense. What is happening? We are overfitting the data. Those words that don't make sense are capturing just noise in the reviews.
#
# For example, the word `helming` appears in only one review:

idf_pipeline.transform(training_df).\
    select('id', fn.explode('words').alias('word')).\
    where(fn.col('word') == 'helming').\
    join(training_df, 'id').\
    first()

# # Regularization

# One way to prevent overfitting during training is to modify the loss function and penalize weight values that are too large.
#
# There are two major regularization techniques, one based on penalizing the squared value of the weight (called L2 or ridge regularization) and anotherbased on penalizing the absolute value of the weight (called L1 or lasso regularization).
#
# The unregularized logistic regression loss function is:
#
# \begin{equation}
# L_\theta(p(X),Y) = - \left( \sum_i Y_i \log p_\theta(X_i) + (1-Y_i)\log (1-p_\theta(X_i)) \right)
# \end{equation}
#
# where $p_\theta(\cdot)$ is the sigmoid function:
#
# \begin{equation}
# p_\theta(X) = \frac{1}{1+\exp(-(\theta_0 + \sum_{j>0} x_j \theta_j))}
# \end{equation}
#
# If we modify the loss function $L_\theta$ slightly
#
# \begin{equation}
# L_\theta^{\lambda}(p(X),Y) = -\left( \sum_i Y_i \log p_\theta(X_i) + (1-Y_i)\log(1-p_\theta(X_i)) \right) + \lambda \sum_{j>0} \theta_j^2
# \end{equation}
#
# we obtain what is known as L2 regularization.
#
# Notice how we increase the loss function by $\lambda$ times the square of the weights. In practice, this means that __we will think twice about increasing the importance of a feature__. This loss function will prevent the algorithm for fitting certain data points, such as outliers or noise, unless the decrease in loss for the data grants it. Also, notice that the penalization doesn't apply to the bias parameter $\theta_0$.
#
# You can see more clearly the effect of such cost function when $\lambda$ goes to infinity: the features will not be used for predicting and only the bias term will matter! This prevents the algorithm from learning altogether, forcing it to underfit!
#
# One problem with L2 regularization is that all weights go to zero uniformly. In a sense, all features will matter but less than with the unregularized loss function. This is a really strange because we do not want all features to matter. In sentiment analysis, we want to select certain features because we want to understand that only some words have effects on the sentiment.
#
# A different modification of the original loss function can achieve this. This regularization is known as L1 or lasso reguarlization and penalizes the _absolute_ value of the weight
#
# \begin{equation}
# L_\theta^{\lambda}(p(X),Y) = -\left( \sum_i Y_i \log p_\theta(X_i) + (1-Y_i)\log(1-p_\theta(X_i)) \right) + \lambda \sum_{j>0} \left| \theta_j \right|
# \end{equation}
#
# The practical effect of L1 regularization is that the difference between a feature having no importance vs some small importance is massively bigger than with L2 regularization. __Therefore, optimizing the L1 loss function usually brings some features to have exactly zero weight.__
#
# One problem with L1 regularization is that it will never select more features that the number of examples. This is because it can always fit the training data perfectly when the number of features equals the number of examples. In our sentimental analysis, this is the case (there are more words than examples).
#
# One way of remedying this is to have a combination of both L1 and L2. This is known as __elastic net regularization__. For this type of regularization, we have to pick a parameter ($\alpha$) deciding to consider L1 vs L2 regularization. If $\alpha=0$, then we choose L2, and if $\alpha=1$ we choose L1. For example, $\alpha=0.5$ means half L1 and half L2.
#
# \begin{equation}
# L_\theta^{\lambda,\alpha}(p(X),Y) = -\left( \sum_i Y_i \log p_\theta(X_i) + (1-Y_i)\log (1-p_\theta(X_i)) \right) + \lambda \left[(1-\alpha) \sum_{j>0} \theta_j^2 + \alpha \sum_{j>0} \left| \theta_j \right| \right]
# \end{equation}
#
# Unfortunately, elastic net regularization comes with two additional parameters, $\lambda$ and $\alpha$, and we must either select them a priori or used the validation set to choose the best one.

# ## Spark allows to fit elatic net regularization easily

lambda_par = 0.02
alpha_par = 0.3
en_lr = LogisticRegression().\
        setLabelCol('score').\
        setFeaturesCol('tfidf').\
        setRegParam(lambda_par).\
        setMaxIter(100).\
        setElasticNetParam(alpha_par)

# And we define a new Pipeline with all steps combined

en_lr_estimator = Pipeline(
    stages=[tokenizer, sw_filter, cv, idf, en_lr])

en_lr_pipeline = en_lr_estimator.fit(training_df)

# Let's look at the performance

en_lr_pipeline.transform(validation_df).select(fn.avg(fn.expr('float(prediction = score)'))).show()

# We improve performance slightly, but whats more important is that we improve the understanding of the word sentiments. Lets take at the weights:

en_weights = en_lr_pipeline.stages[-1].coefficients.toArray()
en_coeffs_df = pd.DataFrame({'word': en_lr_pipeline.stages[2].vocabulary, 'weight': en_weights})

# The most negative words all make sense ("worst" is _actually_ more negative than than "worse")!

en_coeffs_df.sort_values('weight').head(15)

# Same thing with positive words

en_coeffs_df.sort_values('weight', ascending=False).head(15)

# Are there words with _literarily_ zero importance for predicting sentiment? Yes, and most of them!

en_coeffs_df.query('weight == 0.0').shape

# In fact, more than 95% of features are not needed to achieve a __better__ performance than all previous models!

en_coeffs_df.query('weight == 0.0').shape[0]/en_coeffs_df.shape[0]

# Let's look at these _neutral_ words

en_coeffs_df.query('weight == 0.0').head(15)

# But, did we choose the right $\lambda$ and $\alpha$ parameters? We should run an experiment where we try different combinations of them. Fortunately, Spark let us do this by using a grid - a method that generates combination of parameters.

from pyspark.ml.tuning import ParamGridBuilder

# We need to build a new estimator pipeline

en_lr_estimator.getStages()

grid = ParamGridBuilder().\
    addGrid(en_lr.regParam, [0., 0.01, 0.02]).\
    addGrid(en_lr.elasticNetParam, [0., 0.2, 0.4]).\
    build()

# This is the list of parameters that we will try:

grid

all_models = []
for j in range(len(grid)):
    print("Fitting model {}".format(j+1))
    model = en_lr_estimator.fit(training_df, grid[j])
    all_models.append(model)

# estimate the accuracy of each of them:
accuracies = [m.\
    transform(validation_df).\
    select(fn.avg(fn.expr('float(score = prediction)')).alias('accuracy')).\
    first().\
    accuracy for m in all_models]

accuracies

import numpy as np

best_model_idx = np.argmax(accuracies)

# So the best model we found has the following parameters

grid[best_model_idx]

best_model = all_models[best_model_idx]

accuracies[best_model_idx]

# estimate generalization performance
best_model.\
    transform(testing_df).\
    select(fn.avg(fn.expr('float(score = prediction)')).alias('accuracy')).\
    show()

# ## Finally, predicting tweet sentiments

# Now we can use this model to predict sentiments on Twitter

tweets_df = spark.read.parquet('/datasets/tweets.parquet')
tweets_df.show(5, truncate=False)

# We have 1K tweets from each candidate

tweets_df.groupby('handle').agg(fn.count('*')).show()

# We can now predict the sentiment of the Tweet using our best model, we need to rename the column so that it matches our previous pipeline (`review` => ...)

best_model.transform(tweets_df.withColumnRenamed('text', 'review')).select('review', 'prediction').show()

# Now, lets summarize our results in a graph!

# %matplotlib inline

import seaborn

sentiment_pd = best_model.\
    transform(tweets_df.withColumnRenamed('text', 'review')).\
    groupby('handle').\
    agg(fn.avg('prediction').alias('prediction'), 
        (2*fn.stddev('prediction')/fn.sqrt(fn.count('*'))).alias('err')).\
    toPandas()

sentiment_pd.head()

sentiment_pd.plot(x='handle', y='prediction', xerr='err', kind='barh');

# But let's examine some "negative" tweets by Trump

best_model.\
    transform(tweets_df.withColumnRenamed('text', 'review')).\
    where(fn.col('handle') == '@realDonaldTrump').\
    where(fn.col('prediction') == 0).\
    select('review').\
    take(5)

# And Clinton

best_model.\
    transform(tweets_df.withColumnRenamed('text', 'review')).\
    where(fn.col('handle') == '@HillaryClinton').\
    where(fn.col('prediction') == 0).\
    select('review').\
    take(5)

# As you can see, there are lots of room for improvement.

# ## Part 2: Test yourself

# 1. From the IMDB dataframe (`imdb_reviews_df`), compute the average review length between positive and negative reviews. Hint: use the spark sql function `length`. In particular, as we imported the funcions with the name `fn` (using `from pyspark.sql import function as fn`), use `fn.length` with the name of the column.
# 2. In the IMDB review database, are positive reviews longer than negative reviews?
# 3. Using the sentiment dataframe `sentiments_df`, find the imdb reviews with the most number of negative words. __Hint__: You need to tokenize the `review` field in `imdb_review_df` and then join with `sentiments_df`. Finally, perform selection and summary query
# 4. Similar to 3, find the imdb review with the most number of positive words.

# # Part 3: On our own
#
# 1) Using the best model fitted (`best_model`), estimate the generalization error in the testing set (`testing_df`)
#
# 2) One way of analyzing what is wrong with a model is to examine when they fail the hardest. In our case, we could do this by looking at cases in which logistic regression is predicting with high probability a positive sentiment when in fact the actual sentiment is negative. 
#
# To extract the probability of positive sentiment, however, we must extract it from the prediction with a custom function.

from pyspark.ml import feature

# +
from pyspark.sql import types

def probability_positive(probability_column):
    return float(probability_column[1])
func_probability_positive = fn.udf(probability_positive, types.DoubleType())

prediction_probability_df = best_model.transform(validation_df).\
    withColumn('probability_positive', func_probability_positive('probability')).\
    select('id', 'review', 'score', 'probability_positive')
prediction_probability_df.show()
# -

# Analyze the worst predictions that are __very__ wrong and suggest some ways of improving the model. __Hint__: Do a query that would get the highest `probability_positive` values for cases where `score` is `0`, and vice versa.

# 3) Using the best model (`best_model`), predict the sentiment of the following sentences.
#
# a) "Make America great again"
#
# b) "Cats are not always the best companion"
#
# c) "This sentence is not a sentence"

StarcoderdataJupyterScriptsDedupFiltered