ML Task2¶

Goal: Predict the post sentiment.

In [2]:
import findspark
findspark.init()
In [21]:
import pyspark.sql.functions as f
from pyspark.sql.functions import *
from pyspark.sql import SparkSession
from pyspark.sql.types import *
In [4]:
from pyspark.ml.feature import OneHotEncoder, StringIndexer, IndexToString, VectorAssembler
from pyspark.ml.classification import RandomForestClassifier, GBTClassifier, LogisticRegression, LinearSVC, MultilayerPerceptronClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator, RegressionEvaluator
from pyspark.ml import Pipeline, Model
from pyspark.ml.pipeline import PipelineModel
In [5]:
spark = SparkSession.builder.appName("MLTask2").getOrCreate()

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
22/05/04 00:35:07 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
22/05/04 00:35:15 WARN YarnSchedulerBackend$YarnSchedulerEndpoint: Attempted to request executors before the AM has registered!

Make sure your SparkSession is active:

In [6]:
spark
Out[6]:

SparkSession - in-memory

SparkContext

Spark UI

Version
v3.0.0-amzn-0
Master
yarn
AppName
MLTask2

1. Read in the data¶

Create a DataFrame called df_in, which nlp project result.

In [7]:
df_in = spark.read.parquet('s3://yl1269-labdata5/nlp_df/')
In [9]:
df_in.printSchema()

root
 |-- parent_id: string (nullable = true)
 |-- author: string (nullable = true)
 |-- id: string (nullable = true)
 |-- author_premium: boolean (nullable = true)
 |-- author_flair_richtext: string (nullable = true)
 |-- author_flair_text: string (nullable = true)
 |-- author_flair_text_color: string (nullable = true)
 |-- body: string (nullable = true)
 |-- collapsed: boolean (nullable = true)
 |-- controversiality: long (nullable = true)
 |-- no_follow: boolean (nullable = true)
 |-- score: long (nullable = true)
 |-- send_replies: boolean (nullable = true)
 |-- total_awards_received: long (nullable = true)
 |-- stickied: boolean (nullable = true)
 |-- author_created_time: string (nullable = true)
 |-- created_time: string (nullable = true)
 |-- post_date: string (nullable = true)
 |-- is_PtoP: integer (nullable = true)
 |-- is_BuckleUp: integer (nullable = true)
 |-- is_voted: integer (nullable = true)
 |-- author_postcnt: long (nullable = true)
 |-- is_popular_commented: integer (nullable = true)
 |-- is_weekend: integer (nullable = true)
 |-- author_age: integer (nullable = true)
 |-- gme_mentioned: integer (nullable = true)
 |-- sentiment: string (nullable = true)
 |-- blackrock_mentioned: boolean (nullable = true)
 |-- share_mentioned: boolean (nullable = true)
In [10]:
df_stock = spark.read.csv('s3://yl1269-labdata5/GME.csv',header=True)
In [11]:
df_stock.show(10)

+----------+----------+----------+----------+----------+----------+--------+
|      Date|      Open|      High|       Low|     Close| Adj Close|  Volume|
+----------+----------+----------+----------+----------+----------+--------+
|2021-03-01|104.540001|133.990005| 99.970001|120.400002|120.400002|49794000|
|2021-03-02|116.930000|133.199997|112.199997|118.180000|118.180000|33783000|
|2021-03-03|122.510002|127.750000|113.120003|124.180000|124.180000|19273900|
|2021-03-04|125.000000|147.869995|115.300003|132.350006|132.350006|32606900|
|2021-03-05|128.169998|151.529999|127.500000|137.740005|137.740005|30733700|
|2021-03-08|154.889999|210.869995|146.100006|194.500000|194.500000|63565600|
|2021-03-09|217.710007|249.850006|208.509995|246.899994|246.899994|39099300|
|2021-03-10|269.429993|348.500000|172.000000|265.000000|265.000000|71570600|
|2021-03-11|241.639999|281.500000|232.600006|260.000000|260.000000|28312500|
|2021-03-12|275.000000|295.500000|262.269989|264.500000|264.500000|25845900|
+----------+----------+----------+----------+----------+----------+--------+
only showing top 10 rows

2. Clean data to prepare for machine learning¶

In [12]:
#cause no stock data on weekend, only keep weekday df_in data
df = df_in.filter(col("is_weekend")== 0).dropna()
df = df.drop("is_weekend")
In [13]:
#Add usr activity column
df_act = df.groupby("post_date").agg(f.count(col("id")).alias("user_activity"))
df = df.join(df_act, "post_date","left")
In [14]:
#See the column relationship

# adjust figure size and font size
sns.set(rc = {"figure.figsize":(20, 12)})
sns.set(font_scale=1)

# compute the correlation matrix using kendall method
corr = df.limit(100000).toPandas().corr(method='kendall')

mask = np.zeros_like(corr)
mask[np.triu_indices_from(mask)] = True

sns.heatmap(corr,annot=True, fmt='.2f', square=True, mask = mask, cmap="Blues");

/mnt/miniconda/lib/python3.7/site-packages/scipy/stats/stats.py:4812: RuntimeWarning: overflow encountered in long_scalars
  (2 * xtie * ytie) / m + x0 * y0 / (9 * m * (size - 2)))
/mnt/miniconda/lib/python3.7/site-packages/scipy/stats/stats.py:4814: RuntimeWarning: invalid value encountered in sqrt
  np.sqrt(var) / np.sqrt(2)))
In [14]:
df = df.dropna()
In [15]:
df.groupby('sentiment').count().show()

                                                                                

+---------+-------+
|sentiment|  count|
+---------+-------+
| positive|3209651|
|  neutral| 394656|
| negative| 961150|
+---------+-------+
In [17]:
# Compute the weights
n = df.count()
n_sen_p = df.filter(col('sentiment')=='positive').count()
n_sen_n = n - n_sen_p

c = 2
weight_sen_p = n/(2*n_sen_p)
weight_sen_n = n/(2*n_sen_n)
In [18]:
# Assign the weights to a new column
df = df.withColumn('weight_sen', f.when(col('sentiment')=='positive', weight_sen_p).otherwise(weight_sen_n))
In [19]:
#join stock data
df = df.join(df_stock,df.post_date == df_stock.Date,"left")
In [22]:
#change stock data from string to double
df = df.withColumn("Open",col("Open").cast(DoubleType()))\
    .withColumn("High",col("High").cast(DoubleType()))\
    .withColumn("Low",col("Low").cast(DoubleType()))\
    .withColumn("Close",col("Close").cast(DoubleType()))\
    .withColumn("Volume",col("Volume").cast(DoubleType()))\
    .withColumn("no_follow",col("no_follow").cast(IntegerType()))\
In [23]:
df.count()
Out[23]:
4170801
In [ ]:

3. Split data into train, test, and split¶

In [24]:
train_data, test_data, predict_data = df.randomSplit([0.8, 0.18, 0.02], 24)

After splitting into three datasets, report the number of rows for each split.

In [25]:
print("Number of training records: " + str(train_data.count()))
print("Number of testing records : " + str(test_data.count()))
print("Number of prediction records : " + str(predict_data.count()))

                                                                                

Number of training records: 3336530

                                                                                

Number of testing records : 751268

[Stage 76:=====================================================>  (42 + 2) / 44]

Number of prediction records : 83003

4.RandomForestClassifier¶

4.1 Baseline¶

In [26]:
train_data.printSchema()

root
 |-- post_date: string (nullable = true)
 |-- parent_id: string (nullable = true)
 |-- author: string (nullable = true)
 |-- id: string (nullable = true)
 |-- author_premium: boolean (nullable = true)
 |-- author_flair_richtext: string (nullable = true)
 |-- author_flair_text: string (nullable = true)
 |-- author_flair_text_color: string (nullable = true)
 |-- body: string (nullable = true)
 |-- collapsed: boolean (nullable = true)
 |-- controversiality: long (nullable = true)
 |-- no_follow: integer (nullable = true)
 |-- score: long (nullable = true)
 |-- send_replies: boolean (nullable = true)
 |-- total_awards_received: long (nullable = true)
 |-- stickied: boolean (nullable = true)
 |-- author_created_time: string (nullable = true)
 |-- created_time: string (nullable = true)
 |-- is_PtoP: integer (nullable = true)
 |-- is_BuckleUp: integer (nullable = true)
 |-- is_voted: integer (nullable = true)
 |-- author_postcnt: long (nullable = true)
 |-- is_popular_commented: integer (nullable = true)
 |-- author_age: integer (nullable = true)
 |-- gme_mentioned: integer (nullable = true)
 |-- sentiment: string (nullable = true)
 |-- blackrock_mentioned: boolean (nullable = true)
 |-- share_mentioned: boolean (nullable = true)
 |-- user_activity: long (nullable = true)
 |-- weight_sen: double (nullable = false)
 |-- Date: string (nullable = true)
 |-- Open: double (nullable = true)
 |-- High: double (nullable = true)
 |-- Low: double (nullable = true)
 |-- Close: double (nullable = true)
 |-- Adj Close: string (nullable = true)
 |-- Volume: double (nullable = true)
In [27]:
stringIndexer_sentiment = StringIndexer(inputCol="sentiment", outputCol="sentiment_ix")
stringIndexer_author = StringIndexer(inputCol="author", outputCol="author_ix")
stringIndexer_post_date = StringIndexer(inputCol="post_date", outputCol="post_date_ix")

grade_label_fit = stringIndexer_sentiment.fit(df)
grade_label_fit.labels
Out[27]:
['positive', 'negative']
In [28]:
onehot_sentiment= OneHotEncoder(inputCol="sentiment_ix", outputCol="sentiment_vec")
onehot_author= OneHotEncoder(inputCol="author_ix", outputCol="author_vec")
onehot_post_date= OneHotEncoder(inputCol="post_date_ix", outputCol="post_date_vec")
In [29]:
vectorAssembler_features = VectorAssembler(
    inputCols=[ 'collapsed',
                'no_follow',
                'score',
                'total_awards_received',
                'stickied',
                'is_PtoP',
                'is_voted',
                'author_postcnt',
                'author_age',
                'gme_mentioned',
                'share_mentioned',
                'Open',
                'Close'], 
    outputCol= "features")
In [30]:
rf = RandomForestClassifier(labelCol="sentiment_ix", featuresCol="features", numTrees=50, weightCol='weight_sen')
In [31]:
labelConverter = IndexToString(inputCol="prediction", 
                               outputCol="predictedsentiment", 
                               labels= stringIndexer_sentiment.fit(df).labels)
In [32]:
pipeline_rf = Pipeline(stages=[stringIndexer_sentiment,
                               stringIndexer_author,
                               stringIndexer_post_date,
                               onehot_sentiment,
                               onehot_author,
                               onehot_post_date,
                               vectorAssembler_features, 
                               rf,labelConverter])
In [ ]:
model_rf = pipeline_rf.fit(train_data)
In [ ]:
model_rf.transform(train_data)

4.1.1 Model Test Results¶

To evaluate the model, use test data.

In [30]:
predictions_rf = model_rf.transform(test_data)
In [39]:
LABEL = "sentiment_ix"
evaluatorRF = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="accuracy")
accuracy = evaluatorRF.evaluate(predictions_rf)
mse = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="mse").evaluate(predictions_rf)
var = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="var").evaluate(predictions_rf)
recall = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedRecall").evaluate(predictions_rf)
tp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedTruePositiveRate").evaluate(predictions_rf)
fp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedFalsePositiveRate").evaluate(predictions_rf)
In [40]:
precision = tp / (tp + fp)
f1 =  2 * (precision*recall) / (precision + recall)
In [42]:
print("Accuracy = %g" % accuracy)
print("Test Error = %g" % (1.0 - accuracy))
print("Recall = %g" % recall)
print("Precision = %g" % precision)
print("F1 = %g" % f1)
print("MSE = %g" % mse)
print("Var = %g" % var)

Accuracy = 0.703594
Test Error = 0.296406
Recall = 0.703594
Precision = 0.507044
F1 = 0.589364
MSE = 0.557026
Var = 0.189046

4.1.2 Confusion Matrix¶

In [43]:
from sklearn.metrics import confusion_matrix
In [44]:
y_pred=predictions.select("prediction").collect()
y_orig=predictions.select("sentiment_ix").collect()
In [45]:
cm = confusion_matrix(y_orig, y_pred)
print("Confusion Matrix:")
print(cm)

Confusion Matrix:
[[686786      0   8602]
 [204406      0   2195]
 [ 77056      0   6965]]
In [46]:
import pandas as pd
import seaborn as sn
import matplotlib.pyplot as plt

sn.heatmap(cm, annot=True,cmap="Blues")
plt.show()

4.1.3 ROC Curve¶

In this section, we will evaluate the model area under the curve. This requires using the evaluator BinaryClassificationEvaluator, which you can read about in the doc here. To evaluate the model, use test data.

In [70]:
evaluatorRF = BinaryClassificationEvaluator(labelCol="sentiment_ix", rawPredictionCol="prediction", metricName="areaUnderROC")
roc_result = evaluatorRF.evaluate(predictions)
roc_result
Out[70]:
0.5082585803262698

4.2 Tuning¶

In [119]:
stringIndexer_sentiment = StringIndexer(inputCol="sentiment", outputCol="sentiment_ix")
stringIndexer_author = StringIndexer(inputCol="author", outputCol="author_ix")
stringIndexer_post_date = StringIndexer(inputCol="post_date", outputCol="post_date_ix")

grade_label_fit = stringIndexer_sentiment.fit(df)
grade_label_fit.labels
Out[119]:
['positive', 'negative']
In [120]:
onehot_sentiment= OneHotEncoder(inputCol="sentiment_ix", outputCol="sentiment_vec")
onehot_author= OneHotEncoder(inputCol="author_ix", outputCol="author_vec")
onehot_post_date= OneHotEncoder(inputCol="post_date_ix", outputCol="post_date_vec")
In [121]:
vectorAssembler_features = VectorAssembler(
    inputCols=[ 'collapsed',
                'no_follow',
                'score',
                'total_awards_received',
                'stickied',
                'is_PtoP',
                'is_voted',
                'author_postcnt',
                'author_age',
                'gme_mentioned',
                'share_mentioned',
                'Open',
                'High',
                'Low',
                'Close'], 
    outputCol= "features")
In [122]:
rf1 = RandomForestClassifier(labelCol="sentiment_ix", featuresCol="features", numTrees=100, weightCol='weight_sen',maxDepth=10, maxBins=64)
In [123]:
labelConverter = IndexToString(inputCol="prediction", 
                               outputCol="predictedSentiment", 
                               labels= stringIndexer_sentiment.fit(df).labels)
In [124]:
pipeline_rf1 = Pipeline(stages=[stringIndexer_sentiment,
                               stringIndexer_author,
                               stringIndexer_post_date,
                               onehot_sentiment,
                               onehot_author,
                               onehot_post_date,
                               vectorAssembler_features, 
                               rf1,labelConverter])
In [125]:
model_rf1 = pipeline_rf.fit(train_data)
In [126]:
model_rf1.transform(train_data)
Out[126]:
DataFrame[parent_id: string, author: string, id: string, author_premium: boolean, author_flair_richtext: string, author_flair_text: string, author_flair_text_color: string, body: string, collapsed: boolean, controversiality: bigint, no_follow: boolean, score: bigint, send_replies: boolean, total_awards_received: bigint, stickied: boolean, author_created_time: string, created_time: string, post_date: string, is_PtoP: int, is_BuckleUp: int, is_voted: int, author_postcnt: bigint, is_popular_commented: int, is_weekend: int, author_age: int, gme_mentioned: int, sentiment: string, blackrock_mentioned: boolean, share_mentioned: boolean, weight_sen: double, sentiment_ix: double, author_ix: double, post_date_ix: double, sentiment_vec: vector, author_vec: vector, post_date_vec: vector, features: vector, rawPrediction: vector, probability: vector, prediction: double, predictedSentiment: string]

4.1.1 Model Test Results¶

To evaluate the model, use test data.

In [127]:
predictions_rf1 = model_rf.transform(test_data)
In [ ]:
LABEL = "sentiment_ix"
evaluatorRF = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="accuracy")
accuracy = evaluatorRF.evaluate(predictions_rf1)
mse = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="mse").evaluate(predictions_rf1)
var = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="var").evaluate(predictions_rf1)
recall = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedRecall").evaluate(predictions_rf1)
tp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedTruePositiveRate").evaluate(predictions_rf1)
fp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedFalsePositiveRate").evaluate(predictions_rf1)
In [129]:
precision = tp / (tp + fp)
f1 =  2 * (precision*recall) / (precision + recall)
In [130]:
print("Accuracy = %g" % accuracy)
#print("Test Error = %g" % (1.0 - accuracy))
print("Recall = %g" % recall)
print("Precision = %g" % precision)
print("F1 = %g" % f1)
print("MSE = %g" % mse)
print("Var = %g" % var)

Accuracy = 0.765896
Recall = 0.765896
Precision = 0.499846
F1 = 0.60491
MSE = 0.249464
Var = 0.0724037

4.1.2 Confusion Matrix¶

In [131]:
from sklearn.metrics import confusion_matrix
In [136]:
y_pred=predictions_rf1.select("prediction").collect()
y_orig=predictions_rf1.select("sentiment_ix").collect()

22/04/30 10:28:07 WARN DAGScheduler: Broadcasting large task binary with size 6.1 MiB
In [137]:
cm = confusion_matrix(y_orig, y_pred)
print("Confusion Matrix:")
print(cm)

Confusion Matrix:
[[690930     14   4619]
 [205353     29   1213]
 [     0      0      0]]
In [138]:
import pandas as pd
import seaborn as sn
import matplotlib.pyplot as plt

sn.heatmap(cm, annot=True,cmap="Blues")
plt.show()

4.1.3 ROC Curve¶

In [135]:
evaluatorRF = BinaryClassificationEvaluator(labelCol="sentiment_ix", rawPredictionCol="prediction", metricName="areaUnderROC")
roc_result = evaluatorRF.evaluate(predictions_rf1)
roc_result

22/04/30 10:28:01 WARN DAGScheduler: Broadcasting large task binary with size 6.1 MiB
Out[135]:
0.49967507841585135

5.MultilayerPerceptronClassifier¶

5.1 Baseline¶

In [92]:
stringIndexer_sentiment = StringIndexer(inputCol="sentiment", outputCol="sentiment_ix")
stringIndexer_author = StringIndexer(inputCol="author", outputCol="author_ix")
stringIndexer_post_date = StringIndexer(inputCol="post_date", outputCol="post_date_ix")

grade_label_fit = stringIndexer_sentiment.fit(df)
grade_label_fit.labels
Out[92]:
['positive', 'negative']
In [93]:
onehot_sentiment= OneHotEncoder(inputCol="sentiment_ix", outputCol="sentiment_vec")
onehot_author= OneHotEncoder(inputCol="author_ix", outputCol="author_vec")
onehot_post_date= OneHotEncoder(inputCol="post_date_ix", outputCol="post_date_vec")
In [94]:
vectorAssembler_features = VectorAssembler(
    inputCols=[ 'collapsed',
                'no_follow',
                'score',
                'total_awards_received',
                'stickied',
                'is_PtoP',
                'is_voted',
                'author_postcnt',
                'author_age',
                'gme_mentioned',
                'share_mentioned' ], 
    outputCol= "features")
In [95]:
#lr = LogisticRegression(labelCol="sentiment_ix", featuresCol="features", weightCol='weight_sen')
mlp = MultilayerPerceptronClassifier(labelCol="sentiment_ix", featuresCol="features", layers=[11, 2, 2], seed=123)
In [96]:
labelConverter = IndexToString(inputCol="prediction", 
                               outputCol="predictedSentiment", 
                               labels= stringIndexer_sentiment.fit(df).labels)
In [97]:
pipeline_mlp = Pipeline(stages=[stringIndexer_sentiment,
                               stringIndexer_author,
                               stringIndexer_post_date,
                               onehot_sentiment,
                               onehot_author,
                               onehot_post_date,
                               vectorAssembler_features, 
                               mlp,labelConverter])
In [98]:
model_mlp = pipeline_rf.fit(train_data)
In [100]:
model_mlp.transform(train_data)
Out[100]:
DataFrame[parent_id: string, author: string, id: string, author_premium: boolean, author_flair_richtext: string, author_flair_text: string, author_flair_text_color: string, body: string, collapsed: boolean, controversiality: bigint, no_follow: boolean, score: bigint, send_replies: boolean, total_awards_received: bigint, stickied: boolean, author_created_time: string, created_time: string, post_date: string, is_PtoP: int, is_BuckleUp: int, is_voted: int, author_postcnt: bigint, is_popular_commented: int, is_weekend: int, author_age: int, gme_mentioned: int, sentiment: string, blackrock_mentioned: boolean, share_mentioned: boolean, weight_sen: double, sentiment_ix: double, author_ix: double, post_date_ix: double, sentiment_vec: vector, author_vec: vector, post_date_vec: vector, features: vector, rawPrediction: vector, probability: vector, prediction: double, predictedSentiment: string]

Model Test Results¶

To evaluate the model, use test data.

In [101]:
predictions_mlp = model_mlp.transform(test_data)
In [102]:
LABEL = "sentiment_ix"
evaluatorRF = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="accuracy")
accuracy = evaluatorRF.evaluate(predictions_mlp)
mse = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="mse").evaluate(predictions_mlp)
var = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="var").evaluate(predictions_mlp)
recall = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedRecall").evaluate(predictions_mlp)
tp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedTruePositiveRate").evaluate(predictions_mlp)
fp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedFalsePositiveRate").evaluate(predictions_mlp)
In [103]:
precision = tp / (tp + fp)
f1 =  2 * (precision*recall) / (precision + recall)
In [104]:
print("Accuracy = %g" % accuracy)
print("Recall = %g" % recall)
print("Precision = %g" % precision)
print("F1 = %g" % f1)
print("MSE = %g" % mse)
print("Var = %g" % var)

Accuracy = 0.500303
Recall = 0.500303
Precision = 0.519612
F1 = 0.509774
MSE = 0.499697
Var = 0.336474

Confusion Matrix¶

In [105]:
from sklearn.metrics import confusion_matrix
In [110]:
y_pred=predictions_mlp.select("prediction").collect()
y_orig=predictions_mlp.select("sentiment_ix").collect()
In [111]:
cm = confusion_matrix(y_orig, y_pred)
print("Confusion Matrix:")
print(cm)

Confusion Matrix:
[[337071 358492]
 [ 92314 114281]]
In [112]:
import pandas as pd
import seaborn as sn
import matplotlib.pyplot as plt

sn.heatmap(cm, annot=True,cmap="Blues")
plt.show()

ROC Curve¶

In [117]:
evaluatorRF = BinaryClassificationEvaluator(labelCol="sentiment_ix", rawPredictionCol="prediction", metricName="areaUnderROC")
roc_result = evaluatorRF.evaluate(predictions_mlp)
roc_result
Out[117]:
0.5188830430816718

5.2 Tuning¶

In [145]:
#lr = LogisticRegression(labelCol="sentiment_ix", featuresCol="features", weightCol='weight_sen')
mlp1 = MultilayerPerceptronClassifier(labelCol="sentiment_ix", featuresCol="features", layers=[5, 2, 2], seed=123)
In [146]:
labelConverter = IndexToString(inputCol="prediction", 
                               outputCol="predictedSentiment", 
                               labels= stringIndexer_sentiment.fit(df).labels)
In [147]:
pipeline_mlp1 = Pipeline(stages=[stringIndexer_sentiment,
                               stringIndexer_author,
                               stringIndexer_post_date,
                               onehot_sentiment,
                               onehot_author,
                               onehot_post_date,
                               vectorAssembler_features, 
                               mlp2,labelConverter])
In [148]:
model_mlp1 = pipeline_rf.fit(train_data)
In [149]:
model_mlp1.transform(train_data)
Out[149]:
DataFrame[parent_id: string, author: string, id: string, author_premium: boolean, author_flair_richtext: string, author_flair_text: string, author_flair_text_color: string, body: string, collapsed: boolean, controversiality: bigint, no_follow: boolean, score: bigint, send_replies: boolean, total_awards_received: bigint, stickied: boolean, author_created_time: string, created_time: string, post_date: string, is_PtoP: int, is_BuckleUp: int, is_voted: int, author_postcnt: bigint, is_popular_commented: int, is_weekend: int, author_age: int, gme_mentioned: int, sentiment: string, blackrock_mentioned: boolean, share_mentioned: boolean, weight_sen: double, sentiment_ix: double, author_ix: double, post_date_ix: double, sentiment_vec: vector, author_vec: vector, post_date_vec: vector, features: vector, rawPrediction: vector, probability: vector, prediction: double, predictedSentiment: string]

Model Test Results¶

To evaluate the model, use test data.

In [150]:
predictions_mlp1 = model_mlp1.transform(test_data)
In [151]:
LABEL = "sentiment_ix"
evaluatorRF = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="accuracy")
accuracy = evaluatorRF.evaluate(predictions_mlp1)
mse = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="mse").evaluate(predictions_mlp1)
var = RegressionEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="var").evaluate(predictions_mlp1)
recall = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedRecall").evaluate(predictions_mlp1)
tp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedTruePositiveRate").evaluate(predictions_mlp1)
fp = MulticlassClassificationEvaluator(labelCol=LABEL, predictionCol="prediction", metricName="weightedFalsePositiveRate").evaluate(predictions_mlp1)
In [152]:
precision = tp / (tp + fp)
f1 =  2 * (precision*recall) / (precision + recall)
In [153]:
print("Accuracy = %g" % accuracy)
print("Recall = %g" % recall)
print("Precision = %g" % precision)
print("F1 = %g" % f1)
print("MSE = %g" % mse)
print("Var = %g" % var)

Accuracy = 0.500303
Recall = 0.500303
Precision = 0.519612
F1 = 0.509774
MSE = 0.499697
Var = 0.336474

Confusion Matrix¶

In [105]:
from sklearn.metrics import confusion_matrix
In [110]:
y_pred=predictions_mlp1.select("prediction").collect()
y_orig=predictions_mlp1.select("sentiment_ix").collect()
In [154]:
cm = confusion_matrix(y_orig, y_pred)
print("Confusion Matrix:")
print(cm)

Confusion Matrix:
[[690930     14   4619]
 [205353     29   1213]
 [     0      0      0]]
In [155]:
import pandas as pd
import seaborn as sn
import matplotlib.pyplot as plt

sn.heatmap(cm, annot=True,cmap="Blues")
plt.show()

ROC Curve¶

In [156]:
evaluatorRF = BinaryClassificationEvaluator(labelCol="sentiment_ix", rawPredictionCol="prediction", metricName="areaUnderROC")
roc_result = evaluatorRF.evaluate(predictions_mlp)
roc_result
Out[156]:
0.5188830430816718

Save best model to s3¶

In [158]:
model_rf1.write().save('s3://yl1269-labdata5/model_rf1')

22/04/30 10:40:38 WARN TaskSetManager: Stage 1214 contains a task of very large size (1177 KiB). The maximum recommended task size is 1000 KiB.
In [130]:
spark.stop()
In [ ]: