<jupyter_start><jupyter_text><jupyter_code>import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.5-bin-hadoop2.7"<jupyter_output><empty_output><jupyter_text><jupyter_code>import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()<jupyter_output><empty_output><jupyter_text><jupyter_code>!wget https://www.dropbox.com/s/jh8e8mo5x4dg4ia/USvideos.csv #https://www.dropbox.com/s/i6heazqtoo26rvm/youtube_videos.csv
from pyspark.sql.types import *

youtube_schema=StructType([
    StructField("video_id", StringType(), True),
    StructField("trending_date", StringType(), True),
    StructField("title", StringType(), True),
    StructField("channel_title", StringType(), True),
    StructField("category_id", IntegerType(), True),
    StructField("publish_time", StringType(), True),
    StructField("tags", StringType(), True),
    StructField("views", IntegerType(), True),
    StructField("likes", IntegerType(), True),
    StructField("dislikes", IntegerType(), True),
    StructField("comment_count", IntegerType(), True),
    StructField("thumbnail_link", StringType(), True),
    StructField("comments_disabled", StringType(), True),
    StructField("ratings_disabled", StringType(), True),
    StructField("video_error_or_removed", StringType(), True),
    StructField("description", StringType(), True)

])

youtube_schema
df_youtube = spark.read.csv(path='USvideos.csv', sep=',', header = True, schema = youtube_schema,mode="DROPMALFORMED")<jupyter_output><empty_output><jupyter_text><jupyter_code>df_youtube.count()
# Explore data :
# 1. show the first 5 rows
# 2. print schema

df_youtube.show(5,False)
df_youtube.printSchema()


#df_test = df_youtube.select(['category_id', 'likes', 'dislikes', 'comment_count'])
#df_test.show(1)

#show the statistcs of each column
df_youtube.describe().show()
import matplotlib.pyplot as plt

targets = df_youtube.rdd.map(lambda x:float(x['views'])).collect()
#targets = df_youtube.filter(df_youtube.views < 20000000).rdd.map(lambda x:float(x['views'])).collect()


plt.hist(targets, bins=100, color='lightblue', density=True)
fig = plt.gcf()
fig.set_size_inches(16, 10) 
#show correlation between numeric columns

for i in df_youtube.columns[7:11]:
  for j in df_youtube.columns[7:11]:
    if i!=j : print(f'Correlation between  {i} and {j} : %g' % df_youtube.stat.corr(i,j))
from pyspark.mllib.stat import Statistics
import pandas as pd

#We can also compute the correlation matrix and show the heatmap for it

col_names = df_youtube.columns[7:11]
features = df_youtube.rdd.map(lambda row: row[7:11])
corr_mat=Statistics.corr(features, method="pearson")
corr_df = pd.DataFrame(corr_mat)
corr_df.index, corr_df.columns = col_names, col_names

print(corr_df.to_string())
import seaborn as sns

ax = sns.heatmap(
    corr_df, 
    vmin=-1, vmax=1, center=0,
    cmap=sns.diverging_palette(40, 220, n=220)
)
ax.set_xticklabels(
    ax.get_xticklabels(),
    rotation=45,
    horizontalalignment='right'
);
df = df_youtube.select(df_youtube.columns[7:11]).toPandas()
xs = sns.pairplot(df)
from pyspark.ml.feature import VectorAssembler

# Prepare data for linear regression
# For the prediction, we will only use the following variables : category_id, likes, dislikes, comment_count

vectorAssembler = VectorAssembler(inputCols = ['likes', 'dislikes', 'comment_count'], outputCol = 'features')
#vectorAssembler = VectorAssembler(inputCols = ['likes', 'comment_count'], outputCol = 'features')
vdf_youtube = vectorAssembler.transform(df_youtube)

#vdf_youtube.show(1000)

# The linear regression only needs 2 columns : 'features', which is the values of all the independent variables and the target var 'views' 
vdf_youtube = vdf_youtube.select(['features', 'views'])
vdf_youtube.show(5,False)
# Split the data into training and testing set. We use 70% for training and 30% testing 

train_df, test_df = vdf_youtube.randomSplit([0.7, 0.3])
train_df.show(5)
test_df.show(5)
from pyspark.ml.regression import LinearRegression

# Train the model

lr = LinearRegression(featuresCol = 'features', labelCol='views', maxIter=10, regParam=0.3, elasticNetParam=0.8)
lr_model = lr.fit(train_df)
print("Coefficients: " + str(lr_model.coefficients))
print("Intercept: " + str(lr_model.intercept))
# show the model summary 

trainingSummary = lr_model.summary
print("RMSE: %f" % trainingSummary.rootMeanSquaredError)
print("r2: %f" % trainingSummary.r2)

train_df.describe().show()
lr_predictions = lr_model.transform(test_df)
lr_predictions.select("prediction","views","features").show(5)

from pyspark.ml.evaluation import RegressionEvaluator

lr_evaluator = RegressionEvaluator(predictionCol="prediction", \
                 labelCol="views",metricName="r2")
print("R Squared (R2) on test data = %g" % lr_evaluator.evaluate(lr_predictions))
# Predict on test data

test_result = lr_model.evaluate(test_df)
print("Root Mean Squared Error (RMSE) on test data = %g" % test_result.rootMeanSquaredError)
print("numIterations: %d" % trainingSummary.totalIterations)
print("objectiveHistory: %s" % str(trainingSummary.objectiveHistory))
trainingSummary.residuals.show()
res1 = trainingSummary.residuals.toPandas()
ax = sns.distplot(res1)
predictions = lr_model.transform(test_df)
predictions.select("prediction","views","features").show()<jupyter_output>+------------------+--------+--------------+
|        prediction|   views|      features|
+------------------+--------+--------------+
| 253447.8663563669|    2259|     (3,[],[])|
| 253447.8663563669|    4050|     (3,[],[])|
| 253447.8663563669|   34794|     (3,[],[])|
| 253447.8663563669|   36137|     (3,[],[])|
| 253447.8663563669|   36931|     (3,[],[])|
| 253447.8663563669|   85833|     (3,[],[])|
| 253447.8663563669|   90365|     (3,[],[])|
| 253447.8663563669|  110404|     (3,[],[])|
| 253447.8663563669|  322521|     (3,[],[])|
| 253447.8663563669|  593311|     (3,[],[])|
| 253447.8663563669|  839375|     (3,[],[])|
| 253447.8663563669|  904517|     (3,[],[])|
| 253447.8663563669|  905268|     (3,[],[])|
| 253447.8663563669|  912549|     (3,[],[])|
| 253447.8663563669|  921077|     (3,[],[])|
| 253447.8663563669|  947039|     (3,[],[])|
| 253447.8663563669| 3511909|     (3,[],[])|
| 253447.8663563669|20921796|     (3,[],[])|
| 253447.8663563669|56111957|     (3,[],[])|
|252761.38[...]
{'fraction_non_alphanumeric': 0.17255879146550382, 'fraction_numerical': 0.08316461610341068, 'mean_word_length': 5.358415841584159, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 16, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '185145', 'license': 'BSD-2-Clause', 'n_tokens_mistral': 2655, 'n_tokens_neox': 2310, 'n_words': 427, 'path': 'YAVA Essentials_LABS 08_ Machine Learning With Spark.ipynb', 'repo_name': 'project303/DEV122---Yava-Essentials-', 'size': '6421'}