Libraries and Setup¶
In [ ]:
! sudo apt install openjdk-8-jdk
! sudo update-alternatives --config java
In [ ]:
import numpy as np
import pandas as pd
import json
import matplotlib
import matplotlib.pyplot as plt
from matplotlib import cm
from datetime import datetime
import glob
import seaborn as sns
import re
import os
In [ ]:
import boto3
from botocore import UNSIGNED
from botocore.config import Config
s3 = boto3.resource('s3', config=Config(signature_version=UNSIGNED))
s3.Bucket('penn-cis545-files').download_file('youtube_data.zip', 'youtube_data.zip')
!unzip /content/youtube_data.zip
Section 1 : Machine Learning with Sklearn¶
1.1 Data loading and Preprocessing¶
The dataset is a daily record of the top trending YouTube videos.
To determine the year’s top-trending videos, YouTube uses a combination of factors including measuring users interactions (number of views, shares, comments and likes). Top performers on the YouTube trending list are music videos (such as the famously viral “Gangnam Style”), celebrity and/or reality TV performances, and the random dude-with-a-camera viral videos that YouTube is well-known for.
This dataset includes several months (and counting) of data on daily trending YouTube videos. Data is included for numerous countries, with up to 200 listed trending videos per day.
Each region’s data is in a separate file. Data includes:
- Video Title
- Channel title
- Publish time
- Tags
- Views
- Likes
- Dislikes
- Description
- Comment count
The data also includes a category_id field, which varies between regions.
1.1.1: Combining Multiple CSV's¶
There are multiple csv files in the dataset, each corresponding to a specific country. So, the first step is to read them and combine these csv files into a single dataframe.
In [0]:
# Import all the csv files
files = [i for i in glob.glob('/content/youtube_data/*.csv'.format('csv'))]
sorted(files)
Out[0]:
In [0]:
# Combine all into a single dataframe "combined_data" and add a 'country' column.
all_dataframes = list()
country=['CA','FR','IN','US']
i=0
for csv in files:
all_dataframes = all_dataframes+[pd.read_csv(csv,header=[0]).set_index('video_id')]
all_dataframes[i]['country']=country[i]
i=i+1
combined_data = pd.concat(all_dataframes)
In [0]:
combined_data
Out[0]:
1.1.2: Map category Id's to categories¶
In [0]:
# Read the category_id.json file and map the category_id's in the dataframe to the category name.
combined_data['category_id'] = combined_data['category_id'].astype(str)
def find_title(item):
return item['title']
category_df = pd.DataFrame(pd.read_json('/content/youtube_data/US_category_id.json').iloc[:,2].tolist())
category_df['title'] = category_df['snippet'].apply(find_title)
def map_cat(id_str):
return category_df.loc[category_df.id==id_str,'title'].values[0]
combined_data.insert(4, 'category',combined_data['category_id'].apply(map_cat))
In [0]:
combined_data
Out[0]:
1.1.3: Fix datetime format and remove rows with NA's¶
In [0]:
def trending_datetime(time):
return pd.to_datetime('20'+time,format='%Y.%d.%m', errors='ignore')
combined_data['trending_date'] = combined_data['trending_date'].apply(trending_datetime)
combined_data['publish_time'] = pd.to_datetime(combined_data['publish_time'])
# remove NA's
combined_data = combined_data.dropna()
1.2 EDA and Feature Engineering¶
1.2.1: Mean, standard deviation, min and max.¶
In [0]:
# Print some simple statistics like mean, standard deviation, min annd max for each of the numerical features in the dataset.
maxs = combined_data[['views','likes','dislikes','comment_count']].apply(np.max,axis=0)
mins = combined_data[['views','likes','dislikes','comment_count']].apply(np.min,axis=0)
stds = combined_data[['views','likes','dislikes','comment_count']].apply(np.std,axis=0)
means = combined_data[['views','likes','dislikes','comment_count']].apply(np.mean,axis=0)
In [0]:
maxs
Out[0]:
In [0]:
mins
Out[0]:
In [0]:
stds
Out[0]:
In [0]:
means
Out[0]:
1.2.2: Rescale features¶
In [ ]:
# Rescale the likes, views, dislkes and comment_count to log scale (base e) to avoid numerical instability issues
def plusone(num):
return num+1
combined_data['likes_log'] = combined_data['likes'].apply(plusone).apply(np.log)
combined_data['views_log'] = combined_data['views'].apply(np.log)
combined_data['dislikes_log'] = combined_data['dislikes'].apply(plusone).apply(np.log)
combined_data['comment_log'] = combined_data['comment_count'].apply(plusone).apply(np.log)
1.2.3: Plot the distribution¶
In [0]:
sns.distplot(combined_data['likes_log'])
Out[0]:
In [0]:
sns.distplot(combined_data['views_log'])
Out[0]:
In [0]:
sns.distplot(combined_data['dislikes_log'])
Out[0]:
In [0]:
sns.distplot(combined_data['comment_log'])
Out[0]:
1.2.4: Comparing views, likes, dislikes against categories¶
As a next step, we try to gain insights into the data using categories, views, likes and dislikes.
- Number of videos for each category
In [0]:
# Number of videos for each category
plt.figure(figsize=(10,5))
chart = sns.countplot(
data=combined_data,
x='category',
palette='Set1',
order=combined_data['category'].value_counts().index
)
plt.xticks(
rotation=45,
horizontalalignment='right',
fontweight='light',
fontsize='x-large'
)
Out[0]:
- The distribution of category against views
In [0]:
# The distribution of category against views
chart = sns.catplot(
data=combined_data,
x='category',
y='views_log',
kind='box',
palette='Set1'
)
plt.xticks(
rotation=45,
horizontalalignment='right',
fontweight='light'
)
Out[0]:
- The distribution of dislikes against views
In [0]:
# The distribution of dislikes against views
chart = sns.catplot(
data=combined_data,
x='category',
y='dislikes_log',
kind='box',
palette='Set1'
)
plt.xticks(
rotation=45,
horizontalalignment='right',
fontweight='light'
)
Out[0]:
1.2.5: Feature Engineering¶
a. Processing tags¶
In [ ]:
# Count the number of tags.
combined_data["num_tags"]=combined_data['tags'].apply(lambda x:x.count("|")+1)
b. Processing description and title¶
In [ ]:
# Compute the length of description
combined_data["desc_len"]=combined_data['description'].apply(len)
In [ ]:
# Compute the length of title
combined_data["len_title"]=combined_data['title'].apply(len)
c. Processing publish_time.¶
In [ ]:
# Split 'publish_time' feature into three parts time, date, and weekday
combined_data['publish_weekday']= combined_data['publish_time'].apply(lambda x:x.weekday()+1)
combined_data['publish_date'] = combined_data['publish_time'].apply(lambda x:x.date())
combined_data['publish_time'] = combined_data['publish_time'].apply(lambda x:x.time())
d. Number of videos per weekday¶
In [0]:
# Compute the number of videos published per day of the week.
plt.figure(figsize=(10,5))
chart = sns.countplot(
data=combined_data,
x='publish_weekday',
palette='Set1',
order=combined_data['publish_weekday'].value_counts().index
)
1.2.6: Drop all non numeric columns¶
In [0]:
combined_data=combined_data.drop(['title', 'channel_title', 'category', 'tags', 'views', 'likes', 'dislikes', 'comment_count', 'thumbnail_link', 'thumbnail_link', 'description','publish_date', 'publish_time','trending_date'], axis=1)
1.2.7: Convert categorical features in the dataset into one hot vectors.¶
In [0]:
combined_data.publish_weekday = combined_data.publish_weekday.astype('category')
combined_data.country = combined_data.country.astype('category')
combined_data = pd.concat([combined_data, pd.get_dummies(combined_data['publish_weekday'])], axis=1)
combined_data = pd.concat([combined_data, pd.get_dummies(combined_data['country'])], axis=1)
combined_data = pd.concat([combined_data, pd.get_dummies(combined_data['category_id'])], axis=1)
combined_data = combined_data.drop(['category_id','country','publish_weekday'],axis=1)
In [0]:
# Write out the modified data to a file
combined_data_sec_2 = combined_data.copy()
combined_data_sec_2.rename(columns = {'views_log':'label'}, inplace = True)
combined_data_sec_2.to_csv('combined_data.csv')
1.2.8: Split into x and y¶
In [0]:
# Split the data into features and label
combined_data=pd.read_csv('combined_data.csv').set_index('video_id')
In [0]:
label = combined_data['label']
features = combined_data.drop(['label'],axis=1)
1.3 : Machine Learning using sklearn¶
1.3.1 : Split data into train and test¶
In [0]:
from sklearn.model_selection import train_test_split
x_train, x_test, y_train, y_test = train_test_split(features, label, test_size=0.2)
1.3.2: Train Machine Learning Models.¶
1.3.2.1 Linear Regression¶
In [0]:
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
from sklearn.metrics import accuracy_score, mean_absolute_error, mean_squared_error
clf = LinearRegression()
clf.fit(x_train, y_train)
y_pred = clf.predict(x_test)
print("Score:", clf.score(x_test, y_test))
plt.figure(dpi=100)
plt.scatter(y_test, y_pred)
# Different error measures
print("MAE:", mean_absolute_error(y_test, y_pred))
print('MSE:', mean_squared_error(y_test, y_pred))
print('RMSE:', np.sqrt(mean_squared_error(y_test, y_pred)))
1.3.2.2 Dimensionality reduction with PCA¶
Use Principal component analysis to reduce number of dimensions of the dataset
In [0]:
import numpy as np
from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler
# Fit a pca model on your train set
X = StandardScaler().fit_transform(features)
pca = PCA(n_components=38)
X2 = pca.fit_transform(X)
np.set_printoptions(suppress=True)
pca.explained_variance_ratio_
pc_vs_variance = np.cumsum(pca.explained_variance_ratio_)
# Plot the explained_variance_ratio against the number of components
pc_vs_variance
plt.plot(pc_vs_variance)
In [0]:
pca = PCA(n_components=31)
pca.fit(x_train)
x_train_2 = pca.transform(x_train)
1.3.2.3 Random Forest.¶
In [0]:
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import GridSearchCV
# Grid search
param_grid = {
'n_estimators' : [20,30,40],
'max_depth' : [8,12,16]
}
CV_clf = GridSearchCV(estimator=RandomForestRegressor(), param_grid=param_grid,cv=3)
CV_clf.fit(x_train_2, y_train)
CV_clf.best_params_
Out[0]:
In [0]:
# Fit the random forest on the traing data
rfr = RandomForestRegressor(max_depth=16,n_estimators=40)
rfr.fit(x_train_2,y_train)
pca = PCA(n_components=31)
pca.fit(x_test)
x_test_2 = pca.transform(x_test)
# Make predictions
y_pred=rfr.predict(x_test_2)
In [ ]:
np.sqrt(mean_squared_error(y_test, y_pred))
Section 2 : Distributed Machine Learning with Spark¶
Initializing Spark Connection¶
In [ ]:
!apt install libkrb5-dev
!wget https://www-us.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
!tar xf spark-2.4.5-bin-hadoop2.7.tgz
!pip install findspark
!pip install sparkmagic
!pip install pyspark
!pip install pyspark --user
!pip install seaborn --user
!pip install plotly --user
!pip install imageio --user
!pip install folium --user
In [ ]:
!apt update
!apt install gcc python-dev libkrb5-dev
In [0]:
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as F
import os
spark = SparkSession.builder.appName('ml-hw4').getOrCreate()
In [0]:
%load_ext sparkmagic.magics
In [0]:
#graph section
import networkx as nx
# SQLite RDBMS
import sqlite3
# Parallel processing
# import swifter
import pandas as pd
# NoSQL DB
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError, OperationFailure
import os
os.environ['SPARK_HOME'] = '/content/spark-2.4.5-bin-hadoop2.7'
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
import pyspark
from pyspark.sql import SQLContext
In [0]:
try:
if(spark == None):
spark = SparkSession.builder.appName('Initial').getOrCreate()
sqlContext=SQLContext(spark)
except NameError:
spark = SparkSession.builder.appName('Initial').getOrCreate()
sqlContext=SQLContext(spark)
2.1 Data for Spark ML¶
In [0]:
train_sdf = spark.read.format('csv').options(header='true', inferSchema='true').load('combined_data.csv')
In [0]:
train_sdf.show()
In [0]:
#Print the dataframe schema and verify
train_sdf.printSchema()
In [0]:
from pyspark.ml.feature import StringIndexer, VectorAssembler
In [0]:
# Drop unwanted columns
all_columns = train_sdf.columns
drop_columns = ['video_id','label']
columns_to_use = [i for i in all_columns if i not in drop_columns]
In [0]:
# Create a VectorAssembler object
assembler = VectorAssembler(inputCols=columns_to_use, outputCol='features')
In this step, we will create a pipeline with a single stage - the assembler.
In [0]:
from pyspark.ml import Pipeline
# Create a pipeline
pipeline = Pipeline(stages=[assembler])
modified_data_sdf = pipeline.fit(train_sdf).transform(train_sdf).drop(*columns_to_use)
In [0]:
modified_data_sdf.show()
In [0]:
# Split into an 80-20 ratio between the train and test sets.
train_sdf, test_sdf = modified_data_sdf.randomSplit([0.8, 0.2], seed = 2020)
2.2 Linear regression using Spark ML¶
Using Spark ML's linear regression to Train a linear regression model and try to predict the views.
In [0]:
from pyspark.ml.regression import LinearRegression
# Train a linear regression model
lr = LinearRegression(featuresCol = 'features', labelCol='label')
lr_model = lr.fit(train_sdf)
In [0]:
trainingSummary = lr_model.summary
print("RMSE: %f" % trainingSummary.rootMeanSquaredError)
print("r2: %f" % trainingSummary.r2)
In [0]:
predictions = lr_model.transform(test_sdf)
In [0]:
from pyspark.ml.evaluation import RegressionEvaluator
# Compute root mean squared error on the test set
test_result = lr_model.evaluate(test_sdf)
test_rmse_orig = test_result.rootMeanSquaredError
In [0]:
test_rmse_orig
Out[0]:
In [0]:
# Add regularization to avoid overfitting, try out L1, L2 and elastic net
lrl1 = LinearRegression(featuresCol = 'features', labelCol='label',regParam=0.1,elasticNetParam=1.0)
lrl1_model = lrl1.fit(train_sdf)
lrl2 = LinearRegression(featuresCol = 'features', labelCol='label',regParam=0.2,elasticNetParam=0.0)
lrl2_model = lrl2.fit(train_sdf)
lre = LinearRegression(featuresCol = 'features', labelCol='label',regParam=0.1,elasticNetParam=0.5)
lre_model = lre.fit(train_sdf)
In [0]:
# Compute predictions using each of the models
l1_predictions = lrl1_model.transform(test_sdf)
l2_predictions = lrl2_model.transform(test_sdf)
elastic_net_predictions = lre_model.transform(test_sdf)
# Compute root mean squared error on test set for each of your models
test_result_l1 = lrl1_model.evaluate(test_sdf)
test_rmse_l1 = test_result_l1.rootMeanSquaredError
test_result_l2 = lrl2_model.evaluate(test_sdf)
test_rmse_l2 = test_result_l2.rootMeanSquaredError
test_result_e = lre_model.evaluate(test_sdf)
test_rmse_elastic = test_result_e.rootMeanSquaredError
In [0]:
print(test_rmse_l1)
print(test_rmse_l2)
print(test_rmse_elastic)
2.3 Random Forest Regression¶
In [0]:
from pyspark.ml.regression import RandomForestRegressor
# Create a random forest regressor model, fit the training data and evaluate using RegressionEvaluator
rf_reg = RandomForestRegressor(labelCol="label", featuresCol="features")
rf_model = rf_reg.fit(train_sdf)
train_pred = rf_model.transform(train_sdf)
evaluator = RegressionEvaluator(labelCol="label", predictionCol="prediction", metricName="rmse")
train_rmse_rf = evaluator.evaluate(train_pred)#TODO: calculate the training rmse
train_rmse_rf
Out[0]:
In [0]:
# Predictions on the test set
predictions = rf_model.transform(test_sdf)
In [0]:
from pyspark.ml.evaluation import RegressionEvaluator
# Calculate the rmse on the test set
rmse_rf = evaluator.evaluate(predictions)
rmse_rf
Out[0]:
2.4 Dimensionality Reduction using PCA¶
In [0]:
from pyspark.ml.regression import LinearRegression
from pyspark.ml.feature import PCA as PCAml
pca = PCAml(k=31, inputCol='features', outputCol='pcaFeature')
pca_fit = pca.fit(train_sdf)
train_sdf_pca=pca_fit.transform(train_sdf)
pca_model = lr.fit(train_sdf_pca)
pcaSummary = pca_model.summary
training_rmse_pca = pcaSummary.rootMeanSquaredError
training_rmse_pca
In [0]:
training_rmse_pca
Out[0]:
In [0]:
# Your code goes here
from pyspark.ml.evaluation import RegressionEvaluator
test_sdf_pca=pca_fit.transform(test_sdf)
# Get predictions on the test set
predictions_pca = pca_model.transform(test_sdf_pca)
test_result_pca = pca_model.evaluate(test_sdf_pca)
# Get RMSE for test data
test_rmse_pca = test_result_pca.rootMeanSquaredError #TODO: Get RMSE for test data
test_rmse_pca
Out[0]: