<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Download this notebook from https
Step1:  Introduction to dataframes
A DataFrame is a relatively new addition to Spark that stores a distributed dataset of structured columns. It is very similar to an R dataframe or a RDBS table. All columns are of the same type. A DataFrame can be constructed out of a variety of sources, such as a database, CSV files, JSON files, or a Parquet file (columnar storage). The preferred method for storing dataframes is Parquet due to its speed and compression ratio.
Manipulating a DataFrame
We can create a dataframe from a RDD using the sqlContext.
Step2:  From the previous RDDs, we can call the toDF method and specify the name of columns
Step3:  Spark will automatically try to guess the column types. We can take a look at those types
Step4:  Similar to SQL, we can apply a function to a column or several columns.
Step5:  As you can see, the function is not evaluated until an action (e.g., take, show, collect) is taken
Step6:  We can cross (e.g., join) two dataframes ala SQL
Step7:  We can also do outer joins
Step8:  We can apply group functions
Step9:  We can change the name of computed columns
Step10:  Add columns
Step11:  There are many, many types of functions. E.g., see here
Transformers and Estimators
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
Step12:  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.
Step13:  We can now transform the dataframe
Step14:  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.
Step15:  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.
Step16:  Now we need to user the words column that  generated by the tokenizer transformer
Step17:  which results in
Step18:  The column features is a sparse vector representation. For example, for the first document, we have three features present
Step19:  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
Step20:  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
Download the review, sentiment, and tweet datasets
Step21:  Load sentiment data
Step22:  The schema is very simple
Step23:  Lets see how many of each category we have
Step24:  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 reviews that were tagged as positive and negative.
Let's load them
Step25:  Let's take a look at a positive review
Step26:  And a negative one
Step27:  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
Step28:  RegexTokenizer extracts a sequence of matches from the input text. Regular expressions are a powerful tool to extract strings with certain characteristics.
Step29:  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
Step30:  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
Step31:  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.
Step32:  Now if we join that with sentiment, we can see if there are positive and negative words in each review
Step33:  Now we can simply average the sentiment per review id and, say, pick positive when the average is above 0, and negative otherwise.
Step34:  Now, lets compute the accuracy of our prediction
Step35:  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
Step36:  Finally, for this initial Pipeline, we define a counter vectorizer estimator
Step37:  The term frequency vector is represented with a sparse vector. We have 26,384 terms.
Finally, we build another pipeline that takes the output of the previous pipeline and lowers the terms of documents that are very common.
Step38:  Therefore, the idf_pipeline takes the raw text from the datafarme imdb_reviews_df and creates a feature vector vector called tfidf!
Step39:  Data science pipeline for estimating sentiments
First, let's split the data into training, validation, and testing.
Step40:  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 will take the tfidf features and predict whether the review is positive (score == 1) or negative (score == 0).
Step41:  Lets create a pipeline transformation by chaining the idf_pipeline with the logistic regression step (lr)
Step42:  Lets estimate the accuracy
Step43:  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
Step44:  The most negative words are
Step45:  And the most positive
Step46:  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
Step47:  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
Step48:  And we define a new Pipeline
Step49:  Let's look at the performance
Step50:  We improve performance slightly, but whats more important is that we improve the understanding of the word sentiments. Lets take at the weights
Step51:  The most negative words all make sense ("worst" is actually more negative than than "worse")!
Step52:  Same thing with positive words
Step53:  Are there words with literarily zero importance for predicting sentiment? Yes, and most of them!
Step54:  In fact, more than 95% of features are not needed to achieve a better performance than all previous models!
Step55:  Let's look at these neutral words
Step56:  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.
Step57:  We need to build a new estimator pipeline
Step58:  This is the list of parameters that we will try
Step59:  So the best model we found has the following parameters
Step60:  Finally, predicting tweet sentiments
Now we can use this model to predict sentiments on Twitter
Step61:  We have 1K tweets from each candidate
Step62:  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 => ...)
Step63:  Now, lets summarize our results in a graph!
Step64:  But let's examine some "negative" tweets by Trump
Step65:  And Clinton
Step66:  As you can see, there are lots of room for improvement.
Part 2
<ASSISTANT_TASK:>
Python Code:
import findspark
findspark.init()
import pyspark
import numpy as np
conf = pyspark.SparkConf().\
    setAppName('sentiment-analysis').\
    setMaster('local[*]')
from pyspark.sql import SQLContext, HiveContext
sc = pyspark.SparkContext(conf=conf)
sqlContext = HiveContext(sc)
# dataframe functions
from pyspark.sql import functions as fn
Explanation: Download this notebook from https://raw.githubusercontent.com/daniel-acuna/python_data_science_intro/master/notebooks/lab-sentiment_analysis.ipynb
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
End of explanation
# Create a RDDs
documents_rdd = sc.parallelize([
        [1, 'cats are cute', 0],
        [2, 'dogs are playfull', 0],
        [3, 'lions are big', 1],
        [4, 'cars are fast', 1]])
users_rdd = sc.parallelize([
        [0, 'Alice', 20],
        [1, 'Bob', 23],
        [2, 'Charles', 32]])
Explanation: Introduction to dataframes
A DataFrame is a relatively new addition to Spark that stores a distributed dataset of structured columns. It is very similar to an R dataframe or a RDBS table. All columns are of the same type. A DataFrame can be constructed out of a variety of sources, such as a database, CSV files, JSON files, or a Parquet file (columnar storage). The preferred method for storing dataframes is Parquet due to its speed and compression ratio.
Manipulating a DataFrame
We can create a dataframe from a RDD using the sqlContext.
End of explanation
documents_df = documents_rdd.toDF(['doc_id', 'text', 'user_id'])
users_df = users_rdd.toDF(['user_id', 'name', 'age'])
Explanation: From the previous RDDs, we can call the toDF method and specify the name of columns:
End of explanation
documents_df.printSchema()
users_df.printSchema()
Explanation: Spark will automatically try to guess the column types. We can take a look at those types:
End of explanation
from pyspark.sql import functions as fn
# compute the average age of users
user_age_df = users_df.select(fn.avg('age'))
user_age_df
Explanation: Similar to SQL, we can apply a function to a column or several columns.
End of explanation
user_age_df.show()
Explanation: As you can see, the function is not evaluated until an action (e.g., take, show, collect) is taken
End of explanation
users_df.join(documents_df, on='user_id').show()
Explanation: We can cross (e.g., join) two dataframes ala SQL
End of explanation
users_df.join(documents_df, on='user_id', how='left').show()
Explanation: We can also do outer joins
End of explanation
users_df.join(documents_df, 'user_id', how='left').\
    groupby('user_id', 'name').\
    agg(fn.count('text')).\
    show()
Explanation: We can apply group functions
End of explanation
users_df.join(documents_df, 'user_id', how='left').\
    groupby('user_id', 'name').\
    agg(fn.count('text').alias('n_pets')).\
    show()
Explanation: We can change the name of computed columns:
End of explanation
users_df.withColumn('name_length', fn.length('name')).show()
Explanation: Add columns:
End of explanation
from pyspark.ml.feature import Tokenizer
Explanation: There are many, many types of functions. E.g., see here
Transformers and Estimators
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.
End of explanation
# the tokenizer object
tokenizer = Tokenizer().setInputCol('text').setOutputCol('words')
Explanation: 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.
End of explanation
tokenizer.transform(documents_df).show()
Explanation: We can now transform the dataframe
End of explanation
from pyspark.ml.feature import CountVectorizer
Explanation: 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.
End of explanation
count_vectorizer_estimator = CountVectorizer().setInputCol('words').setOutputCol('features')
Explanation: 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.
End of explanation
count_vectorizer_transformer = count_vectorizer_estimator.fit(tokenizer.transform(documents_df))
Explanation: Now we need to user the words column that  generated by the tokenizer transformer
End of explanation
count_vectorizer_transformer.transform(tokenizer.transform(documents_df)).show(truncate=False)
Explanation: which results in:
End of explanation
# list of words in the vocabulary
count_vectorizer_transformer.vocabulary
np.array(count_vectorizer_transformer.vocabulary)[[0, 3, 5]]
Explanation: 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:
End of explanation
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()
Explanation: 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
End of explanation
!wget https://github.com/daniel-acuna/python_data_science_intro/blob/master/data/imdb_reviews_preprocessed.parquet.zip?raw=true -O imdb_reviews_preprocessed.parquet.zip && unzip imdb_reviews_preprocessed.parquet.zip && rm imdb_reviews_preprocessed.parquet.zip
!wget https://github.com/daniel-acuna/python_data_science_intro/blob/master/data/sentiments.parquet.zip?raw=true -O sentiments.parquet.zip && unzip sentiments.parquet.zip && rm sentiments.parquet.zip
!wget https://github.com/daniel-acuna/python_data_science_intro/blob/master/data/tweets.parquet.zip?raw=true -O tweets.parquet.zip && unzip tweets.parquet.zip && rm tweets.parquet.zip
Explanation: 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
Download the review, sentiment, and tweet datasets
End of explanation
sentiments_df = sqlContext.read.parquet('sentiments.parquet')
sentiments_df.printSchema()
Explanation: Load sentiment data
End of explanation
# 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)
Explanation: The schema is very simple: for each word, we have whether it is positive (+1) or negative (-1)
End of explanation
sentiments_df.groupBy('sentiment').agg(fn.count('*')).show()
Explanation: Lets see how many of each category we have
End of explanation
imdb_reviews_df = sqlContext.read.parquet('imdb_reviews_preprocessed.parquet')
Explanation: 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 reviews that were tagged as positive and negative.
Let's load them:
End of explanation
imdb_reviews_df.where(fn.col('score') == 1).first()
Explanation: Let's take a look at a positive review
End of explanation
imdb_reviews_df.where(fn.col('score') == 0).first()
Explanation: And a negative one
End of explanation
from pyspark.ml.feature import RegexTokenizer
Explanation: 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
End of explanation
tokenizer = RegexTokenizer().setGaps(False)\
  .setPattern("\\p{L}+")\
  .setInputCol("review")\
  .setOutputCol("words")
Explanation: RegexTokenizer extracts a sequence of matches from the input text. Regular expressions are a powerful tool to extract strings with certain characteristics.
End of explanation
review_words_df = tokenizer.transform(imdb_reviews_df)
print(review_words_df)
Explanation: 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:
End of explanation
review_words_df.show(5)
Explanation: 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:
End of explanation
review_words_df.select('id', fn.explode('words').alias('word')).show(5)
Explanation: 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.
End of explanation
review_word_sentiment_df = review_words_df.\
    select('id', fn.explode('words').alias('word')).\
    join(sentiments_df, 'word')
review_word_sentiment_df.show(5)
Explanation: Now if we join that with sentiment, we can see if there are positive and negative words in each review:
End of explanation
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)
Explanation: Now we can simply average the sentiment per review id and, say, pick positive when the average is above 0, and negative otherwise.
End of explanation
imdb_reviews_df.\
    join(simple_sentiment_prediction_df, 'id').\
    select(fn.expr('float(score = predicted)').alias('correct')).\
    select(fn.avg('correct')).\
    show()
Explanation: Now, lets compute the accuracy of our prediction
End of explanation
# 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")
Explanation: 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.
End of explanation
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)
Explanation: Finally, for this initial Pipeline, we define a counter vectorizer estimator
End of explanation
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)
Explanation: The term frequency vector is represented with a sparse vector. We have 26,384 terms.
Finally, we build another pipeline that takes the output of the previous pipeline and lowers the terms of documents that are very common.
End of explanation
tfidf_df = idf_pipeline.transform(imdb_reviews_df)
Explanation: Therefore, the idf_pipeline takes the raw text from the datafarme imdb_reviews_df and creates a feature vector vector called tfidf!
End of explanation
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()]
Explanation: Data science pipeline for estimating sentiments
First, let's split the data into training, validation, and testing.
End of explanation
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression().\
    setLabelCol('score').\
    setFeaturesCol('tfidf').\
    setRegParam(0.0).\
    setMaxIter(100).\
    setElasticNetParam(0.)
Explanation: 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 will take the tfidf features and predict whether the review is positive (score == 1) or negative (score == 0).
End of explanation
lr_pipeline = Pipeline(stages=[idf_pipeline, lr]).fit(training_df)
Explanation: Lets create a pipeline transformation by chaining the idf_pipeline with the logistic regression step (lr)
End of explanation
lr_pipeline.transform(validation_df).\
    select(fn.expr('float(prediction = score)').alias('correct')).\
    select(fn.avg('correct')).show()
Explanation: Lets estimate the accuracy:
End of explanation
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})
Explanation: 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:
End of explanation
coeffs_df.sort_values('weight').head(5)
Explanation: The most negative words are:
End of explanation
coeffs_df.sort_values('weight', ascending=False).head(5)
Explanation: And the most positive:
End of explanation
idf_pipeline.transform(training_df).\
    select('id', fn.explode('words').alias('word')).\
    where(fn.col('word') == 'helming').\
    join(training_df, 'id').\
    first()
Explanation: 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:
End of explanation
lambda_par = 0.02
alpha_par = 0.3
en_lr = LogisticRegression().\
        setLabelCol('score').\
        setFeaturesCol('tfidf').\
        setRegParam(lambda_par).\
        setMaxIter(100).\
        setElasticNetParam(alpha_par)
Explanation: 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 p_\theta(X_i) + (1-Y_i)(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 p_\theta(X_i) + (1-Y_i)(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 p_\theta(X_i) + (1-Y_i)(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 p_\theta(X_i) + (1-Y_i)(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
End of explanation
en_lr_pipeline = Pipeline(stages=[idf_pipeline, en_lr]).fit(training_df)
Explanation: And we define a new Pipeline
End of explanation
en_lr_pipeline.transform(validation_df).select(fn.avg(fn.expr('float(prediction = score)'))).show()
Explanation: Let's look at the performance
End of explanation
en_weights = en_lr_pipeline.stages[-1].coefficients.toArray()
en_coeffs_df = pd.DataFrame({'word': vocabulary, 'weight': en_weights})
Explanation: We improve performance slightly, but whats more important is that we improve the understanding of the word sentiments. Lets take at the weights:
End of explanation
en_coeffs_df.sort_values('weight').head(15)
Explanation: The most negative words all make sense ("worst" is actually more negative than than "worse")!
End of explanation
en_coeffs_df.sort_values('weight', ascending=False).head(15)
Explanation: Same thing with positive words
End of explanation
en_coeffs_df.query('weight == 0.0').shape
Explanation: Are there words with literarily zero importance for predicting sentiment? Yes, and most of them!
End of explanation
en_coeffs_df.query('weight == 0.0').shape[0]/en_coeffs_df.shape[0]
Explanation: In fact, more than 95% of features are not needed to achieve a better performance than all previous models!
End of explanation
en_coeffs_df.query('weight == 0.0').head(15)
Explanation: Let's look at these neutral words
End of explanation
from pyspark.ml.tuning import ParamGridBuilder
Explanation: 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.
End of explanation
en_lr_estimator = Pipeline(stages=[idf_pipeline, en_lr])
grid = ParamGridBuilder().\
    addGrid(en_lr.regParam, [0., 0.01, 0.02]).\
    addGrid(en_lr.elasticNetParam, [0., 0.2, 0.4]).\
    build()
Explanation: We need to build a new estimator pipeline
End of explanation
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]
import numpy as np
best_model_idx = np.argmax(accuracies)
Explanation: This is the list of parameters that we will try:
End of explanation
grid[best_model_idx]
best_model = all_models[best_model_idx]
accuracies[best_model_idx]
Explanation: So the best model we found has the following parameters
End of explanation
tweets_df = sqlContext.read.parquet('tweets.parquet')
tweets_df.show(5, truncate=False)
Explanation: Finally, predicting tweet sentiments
Now we can use this model to predict sentiments on Twitter
End of explanation
tweets_df.groupby('handle').agg(fn.count('*')).show()
Explanation: We have 1K tweets from each candidate
End of explanation
best_model.transform(tweets_df.withColumnRenamed('text', 'review')).select('review', 'prediction').show()
Explanation: 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 => ...)
End of explanation
%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');
Explanation: Now, lets summarize our results in a graph!
End of explanation
best_model.\
    transform(tweets_df.withColumnRenamed('text', 'review')).\
    where(fn.col('handle') == '@realDonaldTrump').\
    where(fn.col('prediction') == 0).\
    select('review').\
    take(5)
Explanation: But let's examine some "negative" tweets by Trump
End of explanation
best_model.\
    transform(tweets_df.withColumnRenamed('text', 'review')).\
    where(fn.col('handle') == '@HillaryClinton').\
    where(fn.col('prediction') == 0).\
    select('review').\
    take(5)
Explanation: And Clinton
End of explanation
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()
Explanation: As you can see, there are lots of room for improvement.
Part 2: Test yourself
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.
In the IMDB review database, are positive reviews longer than negative reviews?
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
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.
End of explanation
<END_TASK>