
import pandas as pd
spotify = pd.read_csv("../../data/taylor_swift_spotify.csv",index_col = 0)
spotify.head(3)


spotify = spotify.drop(columns = ['track_number','id','uri'])


# Check datatypes
spotify.info()

# release date is string, not datetime type.
# no missing values in the df, and there 530 records.

<class 'pandas.core.frame.DataFrame'>
Int64Index: 530 entries, 0 to 529
Data columns (total 14 columns):
 #   Column            Non-Null Count  Dtype  
---  ------            --------------  -----  
 0   name              530 non-null    object 
 1   album             530 non-null    object 
 2   release_date      530 non-null    object 
 3   acousticness      530 non-null    float64
 4   danceability      530 non-null    float64
 5   energy            530 non-null    float64
 6   instrumentalness  530 non-null    float64
 7   liveness          530 non-null    float64
 8   loudness          530 non-null    float64
 9   speechiness       530 non-null    float64
 10  tempo             530 non-null    float64
 11  valence           530 non-null    float64
 12  popularity        530 non-null    int64  
 13  duration_ms       530 non-null    int64  
dtypes: float64(9), int64(2), object(3)
memory usage: 62.1+ KB


# Change datatype
spotify['release_date'] = spotify['release_date'].apply(pd.to_datetime)
spotify.info()

<class 'pandas.core.frame.DataFrame'>
Int64Index: 530 entries, 0 to 529
Data columns (total 14 columns):
 #   Column            Non-Null Count  Dtype         
---  ------            --------------  -----         
 0   name              530 non-null    object        
 1   album             530 non-null    object        
 2   release_date      530 non-null    datetime64[ns]
 3   acousticness      530 non-null    float64       
 4   danceability      530 non-null    float64       
 5   energy            530 non-null    float64       
 6   instrumentalness  530 non-null    float64       
 7   liveness          530 non-null    float64       
 8   loudness          530 non-null    float64       
 9   speechiness       530 non-null    float64       
 10  tempo             530 non-null    float64       
 11  valence           530 non-null    float64       
 12  popularity        530 non-null    int64         
 13  duration_ms       530 non-null    int64         
dtypes: datetime64[ns](1), float64(9), int64(2), object(2)
memory usage: 62.1+ KB


import re
# remove the albums released after the reddit data
spotify = spotify[(spotify['release_date']<'2023-03-31')]

def album_cleaning(input_string):
    input_string = re.sub(r'\([^)]*\)', '', input_string) # matches parentheses
    input_string = re.sub(r'\[[^\]]*\]', '', input_string) # matches brackets
    input_string = input_string.split(':')[0] # remove strings after colon
    input_string = input_string.strip() # remove spaces at the begining and the end of the string
    return input_string

# string cleaning on album names
spotify['album'] = spotify['album'].apply(lambda x: album_cleaning(x))

spotify.head(5)


# summary on mean values of acousticness, ..., popularity, duration_ms
summary_table_mean = spotify.groupby(['album']).mean()
summary_table_mean.reset_index(inplace = True)
summary_table_mean.head(5)


summary_table_mean.to_csv("spotify_summary_table.csv",index = False)


# from datetime import timedelta,datetime

# def ms_to_hhmmss(input_ms):
#     d = timedelta(milliseconds = input_ms) # convert milliseconds to hh:mm:ss
    
#     try: # case that the returning time object is in the format of hh:mm:ss.ffffff
#         hhmmss = datetime.strptime(str(d),'%H:%M:%S.%f').time().strftime('%H:%M:%S') # drop the floats
#         hhmmss = datetime.strptime(hhmmss,'%H:%M:%S').time() # get time, exclude date
#         return hhmmss
    
#     except: # case thet the returning time object is is the format of hh:mm:ss
#         hhmmss = datetime.strptime(str(d),'%H:%M:%S').time() # get time, exclude date
#         return hhmmss
    
# summary_table_mean['duration_hhmmss'] = summary_table_mean['duration_ms'].apply(lambda x: ms_to_hhmmss(x))
# summary_table_mean.drop('duration_ms',axis = 1, inplace = True)
# summary_table_mean.head(5)


# Setup - Run only once per Kernel App
%conda install openjdk -y

# install PySpark
%pip install pyspark==3.3.0

# restart kernel
from IPython.core.display import HTML
HTML("<script>Jupyter.notebook.kernel.restart()</script>")

Collecting package metadata (current_repodata.json): done
Solving environment: done


==> WARNING: A newer version of conda exists. <==
  current version: 23.3.1
  latest version: 23.11.0

Please update conda by running

    $ conda update -n base -c defaults conda

Or to minimize the number of packages updated during conda update use

     conda install conda=23.11.0


# All requested packages already installed.


Note: you may need to restart the kernel to use updated packages.
Requirement already satisfied: pyspark==3.3.0 in /opt/conda/lib/python3.10/site-packages (3.3.0)
Requirement already satisfied: py4j==0.10.9.5 in /opt/conda/lib/python3.10/site-packages (from pyspark==3.3.0) (0.10.9.5)
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip
Note: you may need to restart the kernel to use updated packages.


# Import pyspark and build Spark session
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("PySparkApp")
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.2")
    .config(
        "fs.s3a.aws.credentials.provider",
        "com.amazonaws.auth.ContainerCredentialsProvider",
    )
    .getOrCreate()
)

print(spark.version)

Warning: Ignoring non-Spark config property: fs.s3a.aws.credentials.provider

:: loading settings :: url = jar:file:/opt/conda/lib/python3.10/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml

Ivy Default Cache set to: /root/.ivy2/cache
The jars for the packages stored in: /root/.ivy2/jars
org.apache.hadoop#hadoop-aws added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-4e8676a0-5232-45e1-a9e9-4a8b6f83aba7;1.0
	confs: [default]
	found org.apache.hadoop#hadoop-aws;3.2.2 in central
	found com.amazonaws#aws-java-sdk-bundle;1.11.563 in central
:: resolution report :: resolve 425ms :: artifacts dl 24ms
	:: modules in use:
	com.amazonaws#aws-java-sdk-bundle;1.11.563 from central in [default]
	org.apache.hadoop#hadoop-aws;3.2.2 from central in [default]
	---------------------------------------------------------------------
	|                  |            modules            ||   artifacts   |
	|       conf       | number| search|dwnlded|evicted|| number|dwnlded|
	---------------------------------------------------------------------
	|      default     |   2   |   0   |   0   |   0   ||   2   |   0   |
	---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-4e8676a0-5232-45e1-a9e9-4a8b6f83aba7
	confs: [default]
	0 artifacts copied, 2 already retrieved (0kB/25ms)

23/12/08 22:18:00 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

23/12/08 22:18:02 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
23/12/08 22:18:02 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.
3.3.0


import pyspark.sql.functions as f


submissions = spark.read.parquet("s3a://dsan-6000-group-35/submissions.parquet",header = True)
comments = spark.read.parquet("s3a://dsan-6000-group-35/comments.parquet",header = True)

23/12/08 22:18:06 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties


import nltk
nltk.download('stopwords')

[nltk_data] Downloading package stopwords to /root/nltk_data...
[nltk_data]   Package stopwords is already up-to-date!

True


import pandas as pd
from nltk.corpus import stopwords
stopwords_list = stopwords.words('english')

common_words = pd.DataFrame()

for year in [2021,2022,2023]:
    
    # fiter by year
    submission_year = submissions.filter(f.col('year')==year)
    
    # split text data into list
    text = submission_year.select(f.split(submission_year.text, " ").alias("text"))
    # explode list of words
    words = text.select(f.explode(f.col("text")).alias("word"))
    # lower case
    words_lower = words.select(f.lower(f.col("word")).alias("word_lower"))
    
    # remove punctuation
    words_clean = words_lower.select(f.regexp_extract(f.col("word_lower"), "[a-z]+", 0).alias("word"))

    # remove na
    words_clean = words_clean.filter(f.col("word") != "")

    words_count = (words_clean.groupby("word").count().orderBy("count", ascending=False))
    
    top_10 = words_count.limit(10)
    
    top_10 = top_10.toPandas()
    
    common_words[f"{year}_submissions"] =top_10['word']


common_words


for year in [2021,2022,2023]:
    
    # fiter by year
    comments_year = comments.filter(f.col('year')==year)
    
    # split text data into list
    text = comments_year.select(f.split(comments_year.body, " ").alias("text"))
    # explode list of words
    words = text.select(f.explode(f.col("text")).alias("word"))
    # lower case
    words_lower = words.select(f.lower(f.col("word")).alias("word_lower"))
    
    # remove punctuation
    words_clean = words_lower.select(f.regexp_extract(f.col("word_lower"), "[a-z]+", 0).alias("word"))

    # remove na
    words_clean = words_clean.filter(f.col("word") != "")

    words_count = (words_clean.groupby("word").count().orderBy("count", ascending=False))
    
    top_10 = words_count.limit(10)
    
    top_10 = top_10.toPandas()
    
    common_words[f"{year}_comments"] =top_10['word']


common_words


!pip install kaleido

Collecting kaleido
  Using cached kaleido-0.2.1-py2.py3-none-manylinux1_x86_64.whl (79.9 MB)
Installing collected packages: kaleido
Successfully installed kaleido-0.2.1
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip


import plotly.graph_objects as go
import kaleido

fig = go.Figure(data=[go.Table(
    header = dict(values = list(common_words.columns)),
    cells = dict(values = [common_words['2021_submissions'],common_words['2022_submissions'],common_words['2023_submissions'],common_words['2021_comments'],common_words['2022_comments'],common_words['2023_comments']
            ]))])
fig.update_layout(title_text = "Table of Most Common Words over Years")

fig.write_image("yt560_common_words_table.svg")

fig.show()


submissions = submissions.withColumn("text_length", f.length("text"))
comments = comments.withColumn("text_length", f.length("body"))

n = 10000
submissions_sample = submissions.sample(False, n/submissions.count()) 
comments_sample = comments.sample(False,n/comments.count())

submissions_sample = submissions_sample.select('text_length').toPandas()
comments_sample = comments_sample.select('text_length').toPandas()


import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))

# Histogram for submissions
plt.subplot(1, 2, 1)
plt.hist(submissions_sample["text_length"])
plt.title('Text Length Distribution for Submissions Data')
plt.xlabel('Text Length')
plt.ylabel('Frequency')

# Histogram for comments
plt.subplot(1, 2, 2)
plt.hist(comments_sample["text_length"])
plt.title('Text Length Distribution for Comments Data')
plt.xlabel('Text Length')
plt.ylabel('Frequency')

plt.tight_layout()

plt.savefig('text_length_distribution.png')
plt.show()


from pyspark.ml import Pipeline
from pyspark.ml.feature import CountVectorizer
from pyspark.ml.feature import HashingTF, IDF, Tokenizer, RegexTokenizer
from pyspark.ml.feature import StopWordsRemover


submissions = submissions.withColumn("label",f.monotonically_increasing_id())
tokenizer = Tokenizer(inputCol="text", outputCol="words")

vectorizer = CountVectorizer(inputCol="words", outputCol="rawFeatures") 
idf = IDF(inputCol="rawFeatures", outputCol="features", minDocFreq=2)
pipeline = Pipeline(stages=[tokenizer,vectorizer, idf])
model = pipeline.fit(submissions)
result = model.transform(submissions)

23/12/08 22:20:23 WARN DAGScheduler: Broadcasting large task binary with size 3.8 MiB


# Get data under Taylor Swift Subreddit
ts_submissions = submissions.filter("subreddit = 'TaylorSwift'")
ts_comments = comments.filter("subreddit = 'TaylorSwift'")


# Get data under music subreddits
music_submissions = submissions.filter("subreddit != 'TaylorSwift'")
music_comments = comments.filter("subreddit != 'TaylorSwift'")


# Get unique album names, remove Taylor Swift, may cause confusing
import numpy as np
albums = spotify['album'].unique()
albums = np.delete(albums,albums == 'Taylor Swift')
albums = '|'.join(albums)
albums

'Midnights|Red|Fearless|evermore|folklore|Lover|reputation|reputation Stadium Tour Surprise Song Playlist|1989|Speak Now World Tour Live|Speak Now|Fearless Platinum Edition|Live From Clear Channel Stripped 2008'


# apply a regex search to check if the album mentioned in the text, if matched, return the album mentioned, else return None

ts_submissions = ts_submissions.withColumn("album",f.when(f.col("text").rlike(albums),f.regexp_extract(f.col("text"),albums,0)).otherwise(None))
ts_submissions.show(5)

+----+-----+-----------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+-----+---------+
|year|month|  subreddit|    id|              author|        created_utc|                text|num_comments|num_crossposts|score|is_self|stickied|label|    album|
+----+-----+-----------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+-----+---------+
|2021|    3|TaylorSwift|m2tbw3|         storm131713|2021-03-11 15:52:54|The most connecti...|          30|             0|   43|   true|   false|   26| evermore|
|2021|    3|TaylorSwift|m2td8l|      nostalgia-geek|2021-03-11 15:54:30|I just moved and ...|           9|             0|  165|  false|   false|   29|     null|
|2021|    1|TaylorSwift|kua4oj|    franklintbassett|2021-01-10 07:37:10|If you had to say...|          19|             0|   28|   true|   false|   78|Speak Now|
|2021|    1|TaylorSwift|kuaero|stillwantthekidsmenu|2021-01-10 08:00:14|Favorite line fro...|          25|             0|   51|   true|   false|   98| evermore|
|2021|   11|TaylorSwift|qxjbc8|    ScheduleHuman326|2021-11-19 16:03:53|When you are at s...|           1|             0|    1|  false|   false|  156|     null|
+----+-----+-----------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+-----+---------+
only showing top 5 rows


ts_comments = ts_comments.withColumn("album",f.when(f.col("body").rlike(albums),f.regexp_extract(f.col("body"),albums,0)).otherwise(None))
ts_comments.show(5)

[Stage 17:>                                                         (0 + 1) / 1]

+----+-----+-----------+-------+--------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
|year|month|  subreddit|     id|        author|  link_id| parent_id|        created_utc|                body|score|gilded|controversiality|    album|
+----+-----+-----------+-------+--------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
|2022|   10|TaylorSwift|irf7cfk|realscoutfinch|t3_xwui3g|t1_iracq55|2022-10-07 16:46:13|This this this! I...|    1|     0|               0|     null|
|2022|   10|TaylorSwift|irf7gn3|   dietrichs90|t3_xt3tob| t3_xt3tob|2022-10-07 16:47:05|Taylors caption t...|   21|     0|               0|Midnights|
|2022|   10|TaylorSwift|irf7jgy|  evergreenkat|t3_xxy08f| t3_xxy08f|2022-10-07 16:47:40|The whole left co...|    2|     0|               0|     null|
|2022|   10|TaylorSwift|irf7qr0|   sapphicsato|t3_xxr3aj| t3_xxr3aj|2022-10-07 16:49:08|IWAASPIWTWWGROMBF...|    1|     0|               0|     null|
|2022|   10|TaylorSwift|irf7uk8|  hannahberrie|t3_xtuyhw|t1_irc8u25|2022-10-07 16:49:54|               Whelp|    1|     0|               0|     null|
+----+-----+-----------+-------+--------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
only showing top 5 rows


# get non-null rows only
ts_submissions = ts_submissions.filter(f.col('album').isNotNull())
ts_comments = ts_comments.filter(f.col('album').isNotNull())


# for other 2 subreddit data, filter the text mentioned Taylor Swift first
music_submissions = music_submissions.filter(f.col('text').rlike('\W*(Taylor Swift)\W*'))
music_comments = music_comments.filter(f.col('body').rlike('\W*(Taylor Swift)\W*'))

# Then get the album data
music_submissions = music_submissions.withColumn("album",f.when(f.col("text").rlike(albums),f.regexp_extract(f.col("text"),albums,0)).otherwise(None))


music_comments = music_comments.withColumn("album",f.when(f.col("body").rlike(albums),f.regexp_extract(f.col("body"),albums,0)).otherwise(None))

# filter the album related text

music_submissions = music_submissions.filter(f.col('album').isNotNull())
music_comments = music_comments.filter(f.col('album').isNotNull())


# Do a merge with album data and comments/submissions data

summary_table_mean = spark.createDataFrame(summary_table_mean) # pandas df to pyspark df

df1 = ts_submissions.join(summary_table_mean,'album')
df2 = ts_comments.join(summary_table_mean,'album')

# join tables

df3 = music_submissions.join(summary_table_mean,'album')
df4 = music_comments.join(summary_table_mean,'album')


df1.cache()
df2.cache()
df3.cache()
df4.cache()

DataFrame[album: string, year: int, month: int, subreddit: string, id: string, author: string, link_id: string, parent_id: string, created_utc: timestamp, body: string, score: bigint, gilded: bigint, controversiality: bigint, acousticness: double, danceability: double, energy: double, instrumentalness: double, liveness: double, loudness: double, speechiness: double, tempo: double, valence: double, popularity: double, duration_ms: double]


!pip install kaleido

Requirement already satisfied: kaleido in /opt/conda/lib/python3.10/site-packages (0.2.1)
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip


import plotly.express as px

n = 10000
df1_sample = df1.sample(False,n/df1.count())

df1_eda = df1_sample.groupBy('album').count().join(summary_table_mean,'album').select('album','count','popularity')

# apply a scaling
maximum = df1_eda.agg({'count':'max'}).collect()[0][0]
df1_eda = df1_eda.withColumn('count_scaled',f.col('count')/maximum * 100)

df1_eda = df1_eda.withColumn("count_scaled", f.round("count_scaled", 2))\
.withColumn("popularity", f.round("popularity", 2))\
.withColumn("source",f.lit("submissions"))


df2_sample = df2.sample(False,n/df2.count())
df2_eda = df2_sample.groupBy('album').count().join(summary_table_mean,'album').select('album','count','popularity')

# apply a scaling
maximum = df2_eda.agg({'count':'max'}).collect()[0][0]
df2_eda = df2_eda.withColumn('count_scaled',f.col('count')/maximum * 100)

df2_eda = df2_eda.withColumn("count_scaled", f.round("count_scaled", 2)).withColumn("popularity", f.round("popularity", 2)).withColumn("source",f.lit("comments"))


df1_eda = df1_eda.toPandas()
df2_eda = df2_eda.toPandas()
eda_1 = pd.concat([df1_eda,df2_eda])
eda_1


eda_1.to_csv("album_count_and_popularity.csv",index = False)


import kaleido
import plotly.graph_objects as go
from plotly.subplots import make_subplots


fig = px.bar(eda_1,
             x = 'album',
             y = ['count_scaled','popularity'],
             facet_row = 'source',
             title='Grouped Bar Plot of Scaled Album Count and Popularity',
             labels = {'album':'Albums','value':"",'variable':""},
             text_auto = '.2f',
             template = 'plotly_white',
             barmode = 'group',
             width = 1000,
             height = 600)


fig.update_layout(bargap = 0.5)
fig.update_traces(textposition = 'outside')
fig.update_xaxes(tickfont = dict(size = 10),tickangle = 10)
fig.update_yaxes(range=[0,110])

#fig.write_image("yt560_album_count_and_popularity_1.svg")
fig.show()


df3_eda = df3.groupBy('album').count().join(summary_table_mean,'album').select('album','count','popularity')

# apply a scaling
maximum = df3_eda.agg({'count':'max'}).collect()[0][0]
df3_eda = df3_eda.withColumn('count_scaled_submissions',f.col('count')/maximum * 100).withColumnRenamed('count','count_submissions')

df3_eda = df3_eda.withColumn("count_scaled_submissions", f.round("count_scaled_submissions", 2))\
.withColumn("popularity", f.round("popularity", 2))\
.select('album','count_scaled_submissions','count_submissions','popularity')

#df4_sample = df4.sample(False,n/df4.count())

df4_eda = df4.groupBy('album').count().join(summary_table_mean,'album').select('album','count','popularity')

# apply a scaling
maximum = df4_eda.agg({'count':'max'}).collect()[0][0]
df4_eda = df4_eda.withColumn('count_scaled_comments',f.col('count')/maximum * 100).withColumnRenamed('count','count_comments')

df4_eda = df4_eda.withColumn("count_scaled_comments", f.round("count_scaled_comments", 2))\
.withColumn("popularity", f.round("popularity", 2))\
.select('album','count_scaled_comments','count_comments','popularity')

eda_2 = df3_eda.join(df4_eda,['album','popularity'])
eda_2 = eda_2.toPandas()


eda2_table = eda_2[['album','popularity','count_submissions','count_comments']]
eda2_table


eda2_table.to_csv('album_count_popularity_table_music_subs.csv')


fig = px.bar(eda_2,
             x = 'album',
             y = ['count_scaled_submissions','count_scaled_comments','popularity'],
             title='Grouped Bar Plot of Scaled Album Count and Popularity <br><sup>From Both comments data and submissions data under music-related subreddit</sup>',
             labels = {'album':'Albums','value':"",'variable':""},
             text_auto = '.2f',
             template = 'plotly_white',
             barmode = 'group',
             width = 1000,
             height = 450)


fig.update_layout(bargap = 0.5)
fig.update_traces(textposition = 'outside',textfont_size = 300)
fig.update_xaxes(tickfont = dict(size = 10),tickangle = 10)
fig.update_yaxes(range=[0,110])

#fig.write_image("yt560_album_count_and_popularity_2.svg")
fig.show()


# Setup - Run only once per Kernel App
%conda install openjdk -y

# install PySpark
%pip install pyspark==3.3.0

# install spark-nlp
%pip install spark-nlp==5.1.3

# restart kernel
from IPython.core.display import HTML
HTML("<script>Jupyter.notebook.kernel.restart()</script>")

Collecting package metadata (current_repodata.json): done
Solving environment: done


==> WARNING: A newer version of conda exists. <==
  current version: 23.3.1
  latest version: 23.10.0

Please update conda by running

    $ conda update -n base -c defaults conda

Or to minimize the number of packages updated during conda update use

     conda install conda=23.10.0


# All requested packages already installed.


Note: you may need to restart the kernel to use updated packages.
Requirement already satisfied: pyspark==3.3.0 in /opt/conda/lib/python3.10/site-packages (3.3.0)
Requirement already satisfied: py4j==0.10.9.5 in /opt/conda/lib/python3.10/site-packages (from pyspark==3.3.0) (0.10.9.5)
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip
Note: you may need to restart the kernel to use updated packages.
Requirement already satisfied: spark-nlp==5.1.3 in /opt/conda/lib/python3.10/site-packages (5.1.3)
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip
Note: you may need to restart the kernel to use updated packages.


import json
import sparknlp
import numpy as np
import pandas as pd
from sparknlp.base import *
from pyspark.ml import Pipeline
from sparknlp.annotator import *
import pyspark.sql.functions as f
from pyspark.sql import SparkSession
from sparknlp.pretrained import PretrainedPipeline


# Import pyspark and build Spark session
spark = SparkSession.builder \
    .appName("Spark NLP")\
    .master("local[*]")\
    .config("spark.driver.memory","16G")\
    .config("spark.driver.maxResultSize", "0") \
    .config("spark.kryoserializer.buffer.max", "2000M")\
    .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:5.1.3,org.apache.hadoop:hadoop-aws:3.2.2")\
    .config("fs.s3a.aws.credentials.provider","com.amazonaws.auth.ContainerCredentialsProvider")\
    .getOrCreate()

Warning: Ignoring non-Spark config property: fs.s3a.aws.credentials.provider

:: loading settings :: url = jar:file:/opt/conda/lib/python3.10/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml

Ivy Default Cache set to: /root/.ivy2/cache
The jars for the packages stored in: /root/.ivy2/jars
com.johnsnowlabs.nlp#spark-nlp_2.12 added as a dependency
org.apache.hadoop#hadoop-aws added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-3952c4e0-1fb7-48ae-a682-34e5d0002d65;1.0
	confs: [default]
	found com.johnsnowlabs.nlp#spark-nlp_2.12;5.1.3 in central
	found com.typesafe#config;1.4.2 in central
	found org.rocksdb#rocksdbjni;6.29.5 in central
	found com.amazonaws#aws-java-sdk-bundle;1.11.828 in central
	found com.github.universal-automata#liblevenshtein;3.0.0 in central
	found com.google.protobuf#protobuf-java-util;3.0.0-beta-3 in central
	found com.google.protobuf#protobuf-java;3.0.0-beta-3 in central
	found com.google.code.gson#gson;2.3 in central
	found it.unimi.dsi#fastutil;7.0.12 in central
	found org.projectlombok#lombok;1.16.8 in central
	found com.google.cloud#google-cloud-storage;2.20.1 in central
	found com.google.guava#guava;31.1-jre in central
	found com.google.guava#failureaccess;1.0.1 in central
	found com.google.guava#listenablefuture;9999.0-empty-to-avoid-conflict-with-guava in central
	found com.google.errorprone#error_prone_annotations;2.18.0 in central
	found com.google.j2objc#j2objc-annotations;1.3 in central
	found com.google.http-client#google-http-client;1.43.0 in central
	found io.opencensus#opencensus-contrib-http-util;0.31.1 in central
	found com.google.http-client#google-http-client-jackson2;1.43.0 in central
	found com.google.http-client#google-http-client-gson;1.43.0 in central
	found com.google.api-client#google-api-client;2.2.0 in central
	found commons-codec#commons-codec;1.15 in central
	found com.google.oauth-client#google-oauth-client;1.34.1 in central
	found com.google.http-client#google-http-client-apache-v2;1.43.0 in central
	found com.google.apis#google-api-services-storage;v1-rev20220705-2.0.0 in central
	found com.google.code.gson#gson;2.10.1 in central
	found com.google.cloud#google-cloud-core;2.12.0 in central
	found io.grpc#grpc-context;1.53.0 in central
	found com.google.auto.value#auto-value-annotations;1.10.1 in central
	found com.google.auto.value#auto-value;1.10.1 in central
	found javax.annotation#javax.annotation-api;1.3.2 in central
	found commons-logging#commons-logging;1.2 in central
	found com.google.cloud#google-cloud-core-http;2.12.0 in central
	found com.google.http-client#google-http-client-appengine;1.43.0 in central
	found com.google.api#gax-httpjson;0.108.2 in central
	found com.google.cloud#google-cloud-core-grpc;2.12.0 in central
	found io.grpc#grpc-alts;1.53.0 in central
	found io.grpc#grpc-grpclb;1.53.0 in central
	found org.conscrypt#conscrypt-openjdk-uber;2.5.2 in central
	found io.grpc#grpc-auth;1.53.0 in central
	found io.grpc#grpc-protobuf;1.53.0 in central
	found io.grpc#grpc-protobuf-lite;1.53.0 in central
	found io.grpc#grpc-core;1.53.0 in central
	found com.google.api#gax;2.23.2 in central
	found com.google.api#gax-grpc;2.23.2 in central
	found com.google.auth#google-auth-library-credentials;1.16.0 in central
	found com.google.auth#google-auth-library-oauth2-http;1.16.0 in central
	found com.google.api#api-common;2.6.2 in central
	found io.opencensus#opencensus-api;0.31.1 in central
	found com.google.api.grpc#proto-google-iam-v1;1.9.2 in central
	found com.google.protobuf#protobuf-java;3.21.12 in central
	found com.google.protobuf#protobuf-java-util;3.21.12 in central
	found com.google.api.grpc#proto-google-common-protos;2.14.2 in central
	found org.threeten#threetenbp;1.6.5 in central
	found com.google.api.grpc#proto-google-cloud-storage-v2;2.20.1-alpha in central
	found com.google.api.grpc#grpc-google-cloud-storage-v2;2.20.1-alpha in central
	found com.google.api.grpc#gapic-google-cloud-storage-v2;2.20.1-alpha in central
	found com.fasterxml.jackson.core#jackson-core;2.14.2 in central
	found com.google.code.findbugs#jsr305;3.0.2 in central
	found io.grpc#grpc-api;1.53.0 in central
	found io.grpc#grpc-stub;1.53.0 in central
	found org.checkerframework#checker-qual;3.31.0 in central
	found io.perfmark#perfmark-api;0.26.0 in central
	found com.google.android#annotations;4.1.1.4 in central
	found org.codehaus.mojo#animal-sniffer-annotations;1.22 in central
	found io.opencensus#opencensus-proto;0.2.0 in central
	found io.grpc#grpc-services;1.53.0 in central
	found com.google.re2j#re2j;1.6 in central
	found io.grpc#grpc-netty-shaded;1.53.0 in central
	found io.grpc#grpc-googleapis;1.53.0 in central
	found io.grpc#grpc-xds;1.53.0 in central
	found com.navigamez#greex;1.0 in central
	found dk.brics.automaton#automaton;1.11-8 in central
	found com.johnsnowlabs.nlp#tensorflow-cpu_2.12;0.4.4 in central
	found com.microsoft.onnxruntime#onnxruntime;1.15.0 in central
	found org.apache.hadoop#hadoop-aws;3.2.2 in central
:: resolution report :: resolve 4607ms :: artifacts dl 781ms
	:: modules in use:
	com.amazonaws#aws-java-sdk-bundle;1.11.828 from central in [default]
	com.fasterxml.jackson.core#jackson-core;2.14.2 from central in [default]
	com.github.universal-automata#liblevenshtein;3.0.0 from central in [default]
	com.google.android#annotations;4.1.1.4 from central in [default]
	com.google.api#api-common;2.6.2 from central in [default]
	com.google.api#gax;2.23.2 from central in [default]
	com.google.api#gax-grpc;2.23.2 from central in [default]
	com.google.api#gax-httpjson;0.108.2 from central in [default]
	com.google.api-client#google-api-client;2.2.0 from central in [default]
	com.google.api.grpc#gapic-google-cloud-storage-v2;2.20.1-alpha from central in [default]
	com.google.api.grpc#grpc-google-cloud-storage-v2;2.20.1-alpha from central in [default]
	com.google.api.grpc#proto-google-cloud-storage-v2;2.20.1-alpha from central in [default]
	com.google.api.grpc#proto-google-common-protos;2.14.2 from central in [default]
	com.google.api.grpc#proto-google-iam-v1;1.9.2 from central in [default]
	com.google.apis#google-api-services-storage;v1-rev20220705-2.0.0 from central in [default]
	com.google.auth#google-auth-library-credentials;1.16.0 from central in [default]
	com.google.auth#google-auth-library-oauth2-http;1.16.0 from central in [default]
	com.google.auto.value#auto-value;1.10.1 from central in [default]
	com.google.auto.value#auto-value-annotations;1.10.1 from central in [default]
	com.google.cloud#google-cloud-core;2.12.0 from central in [default]
	com.google.cloud#google-cloud-core-grpc;2.12.0 from central in [default]
	com.google.cloud#google-cloud-core-http;2.12.0 from central in [default]
	com.google.cloud#google-cloud-storage;2.20.1 from central in [default]
	com.google.code.findbugs#jsr305;3.0.2 from central in [default]
	com.google.code.gson#gson;2.10.1 from central in [default]
	com.google.errorprone#error_prone_annotations;2.18.0 from central in [default]
	com.google.guava#failureaccess;1.0.1 from central in [default]
	com.google.guava#guava;31.1-jre from central in [default]
	com.google.guava#listenablefuture;9999.0-empty-to-avoid-conflict-with-guava from central in [default]
	com.google.http-client#google-http-client;1.43.0 from central in [default]
	com.google.http-client#google-http-client-apache-v2;1.43.0 from central in [default]
	com.google.http-client#google-http-client-appengine;1.43.0 from central in [default]
	com.google.http-client#google-http-client-gson;1.43.0 from central in [default]
	com.google.http-client#google-http-client-jackson2;1.43.0 from central in [default]
	com.google.j2objc#j2objc-annotations;1.3 from central in [default]
	com.google.oauth-client#google-oauth-client;1.34.1 from central in [default]
	com.google.protobuf#protobuf-java;3.21.12 from central in [default]
	com.google.protobuf#protobuf-java-util;3.21.12 from central in [default]
	com.google.re2j#re2j;1.6 from central in [default]
	com.johnsnowlabs.nlp#spark-nlp_2.12;5.1.3 from central in [default]
	com.johnsnowlabs.nlp#tensorflow-cpu_2.12;0.4.4 from central in [default]
	com.microsoft.onnxruntime#onnxruntime;1.15.0 from central in [default]
	com.navigamez#greex;1.0 from central in [default]
	com.typesafe#config;1.4.2 from central in [default]
	commons-codec#commons-codec;1.15 from central in [default]
	commons-logging#commons-logging;1.2 from central in [default]
	dk.brics.automaton#automaton;1.11-8 from central in [default]
	io.grpc#grpc-alts;1.53.0 from central in [default]
	io.grpc#grpc-api;1.53.0 from central in [default]
	io.grpc#grpc-auth;1.53.0 from central in [default]
	io.grpc#grpc-context;1.53.0 from central in [default]
	io.grpc#grpc-core;1.53.0 from central in [default]
	io.grpc#grpc-googleapis;1.53.0 from central in [default]
	io.grpc#grpc-grpclb;1.53.0 from central in [default]
	io.grpc#grpc-netty-shaded;1.53.0 from central in [default]
	io.grpc#grpc-protobuf;1.53.0 from central in [default]
	io.grpc#grpc-protobuf-lite;1.53.0 from central in [default]
	io.grpc#grpc-services;1.53.0 from central in [default]
	io.grpc#grpc-stub;1.53.0 from central in [default]
	io.grpc#grpc-xds;1.53.0 from central in [default]
	io.opencensus#opencensus-api;0.31.1 from central in [default]
	io.opencensus#opencensus-contrib-http-util;0.31.1 from central in [default]
	io.opencensus#opencensus-proto;0.2.0 from central in [default]
	io.perfmark#perfmark-api;0.26.0 from central in [default]
	it.unimi.dsi#fastutil;7.0.12 from central in [default]
	javax.annotation#javax.annotation-api;1.3.2 from central in [default]
	org.apache.hadoop#hadoop-aws;3.2.2 from central in [default]
	org.checkerframework#checker-qual;3.31.0 from central in [default]
	org.codehaus.mojo#animal-sniffer-annotations;1.22 from central in [default]
	org.conscrypt#conscrypt-openjdk-uber;2.5.2 from central in [default]
	org.projectlombok#lombok;1.16.8 from central in [default]
	org.rocksdb#rocksdbjni;6.29.5 from central in [default]
	org.threeten#threetenbp;1.6.5 from central in [default]
	:: evicted modules:
	com.google.protobuf#protobuf-java-util;3.0.0-beta-3 by [com.google.protobuf#protobuf-java-util;3.21.12] in [default]
	com.google.protobuf#protobuf-java;3.0.0-beta-3 by [com.google.protobuf#protobuf-java;3.21.12] in [default]
	com.google.code.gson#gson;2.3 by [com.google.code.gson#gson;2.10.1] in [default]
	com.amazonaws#aws-java-sdk-bundle;1.11.563 by [com.amazonaws#aws-java-sdk-bundle;1.11.828] in [default]
	---------------------------------------------------------------------
	|                  |            modules            ||   artifacts   |
	|       conf       | number| search|dwnlded|evicted|| number|dwnlded|
	---------------------------------------------------------------------
	|      default     |   77  |   0   |   0   |   4   ||   73  |   0   |
	---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-3952c4e0-1fb7-48ae-a682-34e5d0002d65
	confs: [default]
	0 artifacts copied, 73 already retrieved (0kB/85ms)

23/11/20 11:03:35 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).


print(f"Spark version: {spark.version}")
print(f"sparknlp version: {sparknlp.version()}")

Spark version: 3.3.0
sparknlp version: 5.1.3


submissions = spark.read.parquet("s3a://dsan-6000-group-35/submissions.parquet",header = True)
comments = spark.read.parquet("s3a://dsan-6000-group-35/comments.parquet",header = True)

23/11/20 07:51:28 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties


submissions_sample = submissions.limit(10)


from sparknlp.base import *
from sparknlp.annotator import *
import pyspark.sql.functions as f
from pyspark.ml import Pipeline
from sparknlp.pretrained import PretrainedPipeline
from pyspark.ml.feature import StopWordsRemover

default_stopwords = StopWordsRemover.loadDefaultStopWords('english')

# 1. raw text to document
document_assembler = DocumentAssembler()\
    .setInputCol("text")\
    .setOutputCol("document")

# 2. get the tokens of the text
tokenizer = Tokenizer()\
    .setInputCols(["document"])\
    .setOutputCol("token")

# 3. normalizer
normalizer = Normalizer()\
    .setInputCols(["token"])\
    .setOutputCol("normalized")\
    .setLowercase(True)\
    .setCleanupPatterns(["""[^\w\d\s]"""]) 

# 4. stemmer
stemmer = Stemmer()\
    .setInputCols(["normalized"])\
    .setOutputCol("stem")

# 5. lemmatizer
lemmatizer = LemmatizerModel.pretrained() \
    .setInputCols(["stem"]) \
    .setOutputCol("lemma")

# 6. remove stop words
stopwords_cleaner = StopWordsCleaner()\
    .setInputCols(["lemma"])\
    .setOutputCol("cleanedLemma")\
    .setStopWords(default_stopwords)\
    .setCaseSensitive(False)

# 6. apply a pre trained model
vivekn =  ViveknSentimentModel.pretrained()\
    .setInputCols(["document", "cleanedLemma"])\
    .setOutputCol("result_sentiment")

# 7. remove intermediate outputs, keep sentiment only
finisher = Finisher() \
.setInputCols(["result_sentiment"]) \
.setOutputCols("sentiment")

# Define the pipeline for submissions
nlp_pipeline = Pipeline(stages=[
    document_assembler, 
    tokenizer, 
    normalizer, 
    stemmer,
    lemmatizer, 
    stopwords_cleaner,
    vivekn,
    finisher
])

lemma_antbnc download started this may take some time.

[Stage 11:>                                                         (0 + 4) / 4]

Approximate size to download 907.6 KB
[OK!]
sentiment_vivekn download started this may take some time.
Approximate size to download 873.6 KB
[OK!]


# test on sample data
model = nlp_pipeline.fit(submissions_sample)
result = model.transform(submissions_sample)
result.printSchema()

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.util.SizeEstimator$ (file:/opt/conda/lib/python3.10/site-packages/pyspark/jars/spark-core_2.12-3.3.0.jar) to field java.util.regex.Pattern.pattern
WARNING: Please consider reporting this to the maintainers of org.apache.spark.util.SizeEstimator$
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release

root
 |-- year: integer (nullable = true)
 |-- month: integer (nullable = true)
 |-- subreddit: string (nullable = true)
 |-- id: string (nullable = true)
 |-- author: string (nullable = true)
 |-- created_utc: timestamp (nullable = true)
 |-- text: string (nullable = true)
 |-- num_comments: long (nullable = true)
 |-- num_crossposts: long (nullable = true)
 |-- score: long (nullable = true)
 |-- is_self: boolean (nullable = true)
 |-- stickied: boolean (nullable = true)
 |-- sentiment: array (nullable = true)
 |    |-- element: string (containsNull = true)


# contains na values
result.select('sentiment').show(10)

[Stage 10:>                                                         (0 + 1) / 1]

+----------+
| sentiment|
+----------+
|[positive]|
|[positive]|
|[negative]|
|[negative]|
|      [na]|
|[negative]|
|[negative]|
|      [na]|
|[negative]|
|[negative]|
+----------+


# apply model on submissions data
empty_df = spark.createDataFrame([['']]).toDF("text")
model = nlp_pipeline.fit(empty_df)
submissions_result = model.transform(submissions)

# remove empty returns
submissions_result = submissions_result.filter(f.size(f.col('sentiment')) > 0 )

# convert final outcome from list of elements to string
submissions_result = submissions_result.withColumn("sentiment", f.regexp_replace(f.concat_ws("", f.col("sentiment")), "[\\[\\]]", ""))

# remove nas
submissions_result = submissions_result.filter(f.col('sentiment')!= 'na')
submissions_result.groupBy('sentiment').count().show()

[Stage 15:===========================================>              (3 + 1) / 4]

+---------+------+
|sentiment| count|
+---------+------+
| positive|169538|
| negative|121565|
+---------+------+


submissions_result.show(5)

+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
|year|month|subreddit|    id|              author|        created_utc|                text|num_comments|num_crossposts|score|is_self|stickied|sentiment|
+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
|2021|    6|    Music|o2omjb|     the_dionysian_1|2021-06-18 13:03:18|Bo Burnham - Welc...|        1656|             2|20968|  false|   false| positive|
|2021|    6|    Music|o2onm6|       Seismic_Noise|2021-06-18 13:04:51|LIQUERUS Resident...|           0|             0|    2|  false|   false| positive|
|2021|    6|    Music|o2oop3|SteveNewmanGuitarist|2021-06-18 13:06:31|Steve Newman - Rd...|           0|             0|    1|  false|   false| negative|
|2021|    6|    Music|o2op0p|         Zoinksbeats|2021-06-18 13:07:01|(FREE) Playboi Ca...|           1|             0|    1|  false|   false| negative|
|2021|    6|    Music|o2org3|  BiscuitsAndTeaCups|2021-06-18 13:10:43|Biscuit Sundown -...|           0|             0|    1|  false|   false| negative|
+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
only showing top 5 rows


document_assembler = DocumentAssembler()\
    .setInputCol("body")\
    .setOutputCol("document")

nlp_pipeline = Pipeline(stages=[
    document_assembler, 
    tokenizer, 
    normalizer, 
    stemmer,
    lemmatizer, 
    stopwords_cleaner,
    vivekn,
    finisher
])

empty_df = spark.createDataFrame([['']]).toDF("body")
model = nlp_pipeline.fit(empty_df)
comments_result = model.transform(comments)

comments_result = comments_result.filter(f.size(f.col('sentiment')) > 0 )

# convert final outcome from list of elements to string
comments_result = comments_result.withColumn("sentiment", f.regexp_replace(f.concat_ws("", f.col("sentiment")), "[\\[\\]]", ""))
comments_result = comments_result.filter(f.col('sentiment')!= 'na')


comments_result.show(20)

[Stage 18:>                                                         (0 + 1) / 1]

+----+-----+-----------+-------+--------------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
|year|month|  subreddit|     id|              author|  link_id| parent_id|        created_utc|                body|score|gilded|controversiality|sentiment|
+----+-----+-----------+-------+--------------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
|2022|   10|TaylorSwift|irf7cfk|      realscoutfinch|t3_xwui3g|t1_iracq55|2022-10-07 16:46:13|This this this! I...|    1|     0|               0| positive|
|2022|   10|      Music|irf7cqw|          Alouitious|t3_xxz930|t1_irf6yfd|2022-10-07 16:46:17|4. Once you've bu...|    4|     0|               0| negative|
|2022|   10|      Music|irf7dpm|       ToastedSimian|t3_xy1kr4|t1_ireuzta|2022-10-07 16:46:29|Sadly, I'm a work...|    1|     0|               0| positive|
|2022|   10|      Music|irf7drt|          tommykiddo|t3_xxz930|t1_irf39w8|2022-10-07 16:46:30|Ultimate Guitar h...|    3|     0|               0| negative|
|2022|   10|      Music|irf7ghz|            monkee67|t3_xy0jgc| t3_xy0jgc|2022-10-07 16:47:03|MEH  \n\nBTW this...|   -2|     0|               1| negative|
|2022|   10|TaylorSwift|irf7gn3|         dietrichs90|t3_xt3tob| t3_xt3tob|2022-10-07 16:47:05|Taylors caption t...|   21|     0|               0| positive|
|2022|   10|      Music|irf7jai|          dogsarefun|t3_xxrwic|t1_irdprwc|2022-10-07 16:47:38|Also, Aesop Rock ...|    0|     0|               0| negative|
|2022|   10|TaylorSwift|irf7jgy|        evergreenkat|t3_xxy08f| t3_xxy08f|2022-10-07 16:47:40|The whole left co...|    2|     0|               0| negative|
|2022|   10|      Music|irf7txv|             Kidspud|t3_xy0jgc|t1_irepwh7|2022-10-07 16:49:47|I agree 100%. I f...|   17|     0|               0| negative|
|2022|   10|TaylorSwift|irf7vm7|       AutoModerator|t3_xy42ef| t3_xy42ef|2022-10-07 16:50:07|**All posts are a...|    1|     0|               0| positive|
|2022|   10|      Music|irf813a|         okdude23232|t3_91z3j5|t1_ino5ijx|2022-10-07 16:51:15|people who call t...|    1|     0|               0| positive|
|2022|   10|TaylorSwift|irf8464|              vlarek|t3_xt3tob|t1_irf7rn4|2022-10-07 16:51:53|Grey's aired last...|    7|     0|               0| positive|
|2022|   10|TaylorSwift|irf86gd|         robynnc1290|t3_xt3tob|t1_irf7gn3|2022-10-07 16:52:22|I think there’s s...|    8|     0|               0| negative|
|2022|   10|      Music|irf86z7|          -n0isyb0y-|t3_lzmzv9| t3_lzmzv9|2022-10-07 16:52:29|I am not sure but...|    1|     0|               0| positive|
|2022|   10|      Music|irf8aqn|     drewisawesome14|t3_xx7syx|t1_irdn777|2022-10-07 16:53:16|That’s interestin...|    1|     0|               0| negative|
|2022|   10|      Music|irf8bch|  JoeCorsonStageDeli|t3_xy0jgc| t3_xy0jgc|2022-10-07 16:53:23|Used to really re...|    7|     0|               0| positive|
|2022|   10|TaylorSwift|irf8hei|               lom41|t3_xxy08f|t1_irelrrs|2022-10-07 16:54:38|I love the anti-h...|    5|     0|               0| positive|
|2022|   10|TaylorSwift|irf8ijo|        indievibes23|t3_xxt7wd| t3_xxt7wd|2022-10-07 16:54:52|I’M SO EXCITED! I...|    3|     0|               0| negative|
|2022|   10|TaylorSwift|irf8j4f|OnlyTSwiftSubreddits|t3_xt3tob|t1_ireyy6q|2022-10-07 16:54:59|She's been really...|    2|     0|               0| positive|
|2022|   10|      Music|irf8m64|    Mind-Reflections|t3_xy0jgc| t3_xy0jgc|2022-10-07 16:55:38|This whole album ...|    1|     0|               0| positive|
+----+-----+-----------+-------+--------------------+---------+----------+-------------------+--------------------+-----+------+----------------+---------+
only showing top 20 rows


# save df with sentiment results
bucket = 'dsan-6000-group-35'
submissions_result.write.mode('overwrite').parquet(f"s3a://{bucket}/submissions_with_sentiment.parquet")
comments_result.write.mode('overwrite').parquet(f"s3a://{bucket}/comments_with_sentiment.parquet")


# read comments & submissions with sentiment data
bucket = 'dsan-6000-group-35'
submissions = spark.read.parquet(f"s3a://{bucket}/submissions_with_sentiment.parquet")
comments = spark.read.parquet(f"s3a://{bucket}/comments_with_sentiment.parquet")


submissions_result_summary_table = submissions.groupBy('sentiment').count()
submissions_result_summary_table.show()

[Stage 85:>                                                         (0 + 2) / 2]

+---------+------+
|sentiment| count|
+---------+------+
| positive|169538|
| negative|121565|
+---------+------+


comments_result_summary_table = comments.groupBy('sentiment').count()
comments_result_summary_table.show()

[Stage 88:==================================================>       (7 + 1) / 8]

+---------+-------+
|sentiment|  count|
+---------+-------+
| positive|2892614|
| negative|1963272|
+---------+-------+


submissions.show(5)

+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
|year|month|subreddit|    id|              author|        created_utc|                text|num_comments|num_crossposts|score|is_self|stickied|sentiment|
+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
|2021|    6|    Music|o2omjb|     the_dionysian_1|2021-06-18 13:03:18|Bo Burnham - Welc...|        1656|             2|20968|  false|   false| positive|
|2021|    6|    Music|o2onm6|       Seismic_Noise|2021-06-18 13:04:51|LIQUERUS Resident...|           0|             0|    2|  false|   false| positive|
|2021|    6|    Music|o2oop3|SteveNewmanGuitarist|2021-06-18 13:06:31|Steve Newman - Rd...|           0|             0|    1|  false|   false| negative|
|2021|    6|    Music|o2op0p|         Zoinksbeats|2021-06-18 13:07:01|(FREE) Playboi Ca...|           1|             0|    1|  false|   false| negative|
|2021|    6|    Music|o2org3|  BiscuitsAndTeaCups|2021-06-18 13:10:43|Biscuit Sundown -...|           0|             0|    1|  false|   false| negative|
+----+-----+---------+------+--------------------+-------------------+--------------------+------------+--------------+-----+-------+--------+---------+
only showing top 5 rows


submissions_grouped = submissions.groupBy('year','month','sentiment').count()
submissions_grouped = submissions_grouped.groupBy('year', 'month').pivot('sentiment').agg({'count': 'sum'})
submissions_grouped.show()

[Stage 99:>                                                         (0 + 2) / 2]

+----+-----+--------+--------+
|year|month|negative|positive|
+----+-----+--------+--------+
|2022|   10|    5001|    6790|
|2021|    8|    3994|    5650|
|2021|    6|    5009|    6943|
|2021|    5|    6113|    7996|
|2021|   10|    3436|    5370|
|2022|    2|    3568|    5120|
|2022|    7|    4002|    5682|
|2021|   11|    3904|    6175|
|2021|    9|    3585|    5440|
|2022|   11|    4664|    6475|
|2022|    3|    3996|    5612|
|2023|    3|    4439|    6501|
|2021|   12|    3766|    5750|
|2021|    7|    4079|    5951|
|2023|    2|    3922|    5639|
|2021|    3|    6208|    8088|
|2021|    2|    6041|    8183|
|2022|    1|    3888|    6066|
|2022|    5|    3881|    5420|
|2021|    1|    7401|    8930|
+----+-----+--------+--------+
only showing top 20 rows


!pip install kaleido

Collecting kaleido
  Using cached kaleido-0.2.1-py2.py3-none-manylinux1_x86_64.whl (79.9 MB)
Installing collected packages: kaleido
Successfully installed kaleido-0.2.1
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

[notice] A new release of pip is available: 23.2.1 -> 23.3.1
[notice] To update, run: pip install --upgrade pip


df = submissions_grouped.toPandas()
df['date'] = pd.to_datetime(df[['year', 'month']].assign(DAY=1))


df.to_csv("submissions_sentiment_over_time.csv",index = False)


import plotly.express as px

fig = px.bar(df,
             x='date',
             y=['positive', 'negative'],
             title='Sentiment Over Time for Submissions data',
             labels={'year_month': 'Year-Month', 'value': 'Count', 'variable': 'Sentiment'},
             category_orders={'sentiment': ['positive', 'negative']},
             height=500,
             template = 'plotly_white')
fig.update_layout(bargap = 0.2)
fig.update_xaxes(tickfont = dict(size = 9))

#fig.write_image("yt560_sentiment_over_time_1.svg")

fig.show()


comments_grouped = comments.groupBy('year','month','sentiment').count()
comments_grouped = comments_grouped.groupBy('year', 'month').pivot('sentiment').agg({'count': 'sum'})

df = comments_grouped.toPandas()
df['date'] = pd.to_datetime(df[['year', 'month']].assign(DAY=1))
df.to_csv("comments_sentiment_over_time.csv",index = False)

import plotly.express as px
fig = px.bar(df,
             x='date',
             y=['positive', 'negative'],
             title='Sentiment Over Time for Comments data',
             labels={'year_month': 'Year-Month', 'value': 'Count', 'variable': 'Sentiment'},
             category_orders={'sentiment': ['positive', 'negative']},
             height=500,
             template = 'plotly_white')
fig.update_layout(bargap = 0.2)
fig.update_xaxes(tickfont = dict(size = 9))

#fig.write_image("yt560_sentiment_over_time_2.svg")

fig.show()


import numpy as np
albums = spotify['album'].unique()
albums = np.delete(albums,albums == 'Taylor Swift')
albums = '|'.join(albums)
albums


submissions = submissions.withColumn("album",f.when(f.col("text").rlike(albums),f.regexp_extract(f.col("text"),albums,0)).otherwise(None))
comments = comments.withColumn("album",f.when(f.col("body").rlike(albums),f.regexp_extract(f.col("body"),albums,0)).otherwise(None))

submissions = submissions.filter("subreddit = 'TaylorSwift'")
comments = comments.filter("subreddit = 'TaylorSwift'")

submissions = submissions.filter(f.col('album').isNotNull())
comments = comments.filter(f.col('album').isNotNull())

summary_table_mean = spark.createDataFrame(summary_table_mean) # pandas df to pyspark df

submissions_data = submissions.join(summary_table_mean,'album')
comments_data = comments.join(summary_table_mean,'album')

submissions_data = submissions_data.select('sentiment','album','popularity','energy','valence')
comments_data = comments_data.select('sentiment','album','popularity','energy','valence')


from sklearn.preprocessing import MinMaxScaler
df1 = submissions_data.groupBy('sentiment','album','popularity','energy','valence').count()
df1 = df1.toPandas()
scaler = MinMaxScaler()
df1['scaled_count'] = scaler.fit_transform(df1[['count']])
df1['popularity'] = df1['popularity']/100

df1


df1.to_csv("album_sentiment_table.csv",index = False)


import plotly.express as px
import plotly.graph_objects as go
import kaleido

#df1 = df1.toPandas()
x = df1['album'].unique()
fig = go.Figure(data = [
    go.Bar(name = 'negative',x=df1['album'].unique(),y=df1[df1['sentiment']=='negative']['scaled_count'],marker_color = 'lightslategrey'),
    go.Bar(name = 'positive',x=df1['album'].unique(),y=df1[df1['sentiment']=='positive']['scaled_count'],marker_color='crimson')
])

fig.add_trace(go.Scatter(x=df1['album'].unique(), y=df1[df1['sentiment']=='positive']['popularity'], mode='lines', name='Popularity', line=dict(color='green')))
fig.add_trace(go.Scatter(x=df1['album'].unique(), y=df1[df1['sentiment']=='positive']['energy'], mode='lines', name='Energy', line=dict(color='orange')))
fig.add_trace(go.Scatter(x=df1['album'].unique(), y=df1[df1['sentiment']=='positive']['valence'], mode='lines', name='Valence',line=dict(color='purple')))

fig.update_layout(barmode='group', title='Bar plot with album sentiment and popularity/energy/valence',
                  xaxis_title='Albums', yaxis_title='Scaled Count / Values',
                  height = 600,
                  width = 1000)

#fig.write_image("yt560_album_sentiment_and_popularity_1.svg")
fig.show()


df2 = comments_data.groupBy('sentiment','album','popularity','energy','valence').count()
df2 = df2.toPandas()
scaler = MinMaxScaler()
df2['scaled_count'] = scaler.fit_transform(df2[['count']])
df2['popularity'] = df2['popularity']/100
df2.to_csv("album_sentiment_table_comments.csv",index = False)


x = df2['album'].unique()
neg_data = df2[df2['sentiment']=='negative']
pos_data = df2[df2['sentiment']=='positive']

fig = go.Figure(data = [
    go.Bar(name = 'negative',x=x,y=neg_data['scaled_count'],marker_color ='lightslategrey'),
    go.Bar(name = 'positive',x=x,y=pos_data['scaled_count'],marker_color='crimson')
])

fig.add_trace(go.Scatter(x=x, y=pos_data['popularity'], mode='lines', name='Popularity', line=dict(color='green')))
fig.add_trace(go.Scatter(x=x, y=pos_data['energy'], mode='lines', name='Energy', line=dict(color='orange')))
fig.add_trace(go.Scatter(x=x, y=pos_data['valence'], mode='lines', name='Valence',line=dict(color='purple')))

fig.update_layout(barmode='group', title='Bar plot with album sentiment and popularity/energy/valence',
                  xaxis_title='Albums', yaxis_title='Scaled Count / Values',
                  height = 600,
                  width = 1000)

#fig.write_image("yt560_album_sentiment_and_popularity_2.svg")
fig.show()


import plotly.graph_objects as go
import pandas as pd
tb = pd.read_csv("../../data/csv/nlp/yt560_popularity_table.csv",index_col = 0)
fig = go.Figure(data=[go.Table(
    header = dict(values = list(tb.columns)),
    cells = dict(values = [tb.album,tb.popularity,tb.count_submissions,tb.count_comments])
            )])
fig.update_layout(height = 450,title_text = "Album popularity and count table from music-related subreddit")
#fig.write_image("yt560_popularity_table.svg")
fig.show()


eda_1.to_csv("yt560_popularity_table_2.csv")


eda_1 = pd.read_csv("../../data/csv/nlp/yt560_popularity_table_2.csv",index_col = 0)
eda_1 = eda_1.drop(columns = ['count_scaled'])

fig = go.Figure(data=[go.Table(
    header = dict(values = list(eda_1.columns)),
    cells = dict(values = [eda_1.album,eda_1['count'],eda_1.popularity,eda_1.source])
            )])
fig.update_layout(height = 700,
                  title_text = "Album popularity and count table from Taylor Swift subreddit")
#fig.write_image("yt560_popularity_table_2.svg")
fig.show()


eda_1.columns.drop('count_scaled')

Index(['album', 'count', 'popularity', 'source'], dtype='object')

	name	album	release_date	track_number	id	uri	acousticness	danceability	energy	instrumentalness	liveness	loudness	speechiness	tempo	valence	popularity	duration_ms
0	Welcome To New York (Taylor's Version)	1989 (Taylor's Version) [Deluxe]	2023-10-27	1	4WUepByoeqcedHoYhSNHRt	spotify:track:4WUepByoeqcedHoYhSNHRt	0.009420	0.757	0.610	0.000037	0.3670	-4.840	0.0327	116.998	0.685	80	212600
1	Blank Space (Taylor's Version)	1989 (Taylor's Version) [Deluxe]	2023-10-27	2	0108kcWLnn2HlH2kedi1gn	spotify:track:0108kcWLnn2HlH2kedi1gn	0.088500	0.733	0.733	0.000000	0.1680	-5.376	0.0670	96.057	0.701	80	231833
2	Style (Taylor's Version)	1989 (Taylor's Version) [Deluxe]	2023-10-27	3	3Vpk1hfMAQme8VJ0SNRSkd	spotify:track:3Vpk1hfMAQme8VJ0SNRSkd	0.000421	0.511	0.822	0.019700	0.0899	-4.785	0.0397	94.868	0.305	81	231000

	name	album	release_date	acousticness	danceability	energy	instrumentalness	liveness	loudness	speechiness	tempo	valence	popularity	duration_ms
88	Lavender Haze	Midnights	2022-10-22	0.2040	0.735	0.444	0.001200	0.1700	-10.519	0.0684	97.038	0.0984	72	202395
89	Maroon	Midnights	2022-10-22	0.0593	0.658	0.378	0.000000	0.0976	-8.300	0.0379	108.034	0.0382	71	218270
90	Anti-Hero	Midnights	2022-10-22	0.1330	0.638	0.634	0.000001	0.1520	-6.582	0.0457	96.953	0.5190	72	200690
91	Snow On The Beach (feat. Lana Del Rey)	Midnights	2022-10-22	0.7350	0.659	0.323	0.003210	0.1160	-13.425	0.0436	110.007	0.1540	70	256124
92	You're On Your Own, Kid	Midnights	2022-10-22	0.4160	0.694	0.380	0.000008	0.1260	-10.307	0.0614	120.044	0.3760	72	194206

	album	acousticness	danceability	energy	instrumentalness	liveness	loudness	speechiness	tempo	valence	popularity	duration_ms
0	1989	0.191166	0.639781	0.655000	8.077356e-04	0.183453	-7.238875	0.129922	125.837156	0.457619	68.187500	220422.250000
1	Fearless	0.201353	0.564949	0.638513	9.976923e-07	0.156754	-5.889692	0.036233	125.236641	0.409359	65.948718	246291.384615
2	Fearless Platinum Edition	0.202448	0.575947	0.601053	2.270000e-06	0.159826	-5.708211	0.031905	123.089263	0.380053	45.947368	250874.789474
3	Live From Clear Channel Stripped 2008	0.551250	0.548125	0.598875	0.000000e+00	0.144113	-5.067000	0.045125	137.639000	0.472125	39.375000	209469.375000
4	Lover	0.333743	0.658222	0.545222	7.330572e-04	0.115233	-8.013278	0.099117	119.972722	0.481444	82.611111	206187.833333

	album	count	popularity	count_scaled	source
0	1989	1107	68.19	59.68	submissions
1	Fearless	1206	65.95	65.01	submissions
2	Lover	1174	82.61	63.29	submissions
3	Midnights	1855	76.36	100.00	submissions
4	Red	1769	59.75	95.36	submissions
5	Speak Now	706	55.59	38.06	submissions
6	Speak Now World Tour Live	8	49.00	0.43	submissions
7	evermore	730	73.34	39.35	submissions
8	folklore	958	66.10	51.64	submissions
9	reputation	489	82.93	26.36	submissions
0	1989	1662	68.19	87.75	comments
1	Fearless	1003	65.95	52.96	comments
2	Lover	1451	82.61	76.61	comments
3	Midnights	674	76.36	35.59	comments
4	Red	1894	59.75	100.00	comments
5	Speak Now	762	55.59	40.23	comments
6	evermore	935	73.34	49.37	comments
7	folklore	1082	66.10	57.13	comments
8	reputation	667	82.93	35.22	comments

	sentiment	album	popularity	energy	valence	count	scaled_count
0	negative	1989	0.681875	0.655000	0.457619	327	0.268272
1	positive	1989	0.681875	0.655000	0.457619	707	0.583887
2	negative	Fearless	0.659487	0.638513	0.409359	284	0.232558
3	positive	Fearless	0.659487	0.638513	0.409359	870	0.719269
4	positive	Lover	0.826111	0.545222	0.481444	742	0.612957
5	negative	Lover	0.826111	0.545222	0.481444	307	0.251661
6	positive	Midnights	0.763636	0.435000	0.259288	1014	0.838870
7	negative	Midnights	0.763636	0.435000	0.259288	602	0.496678
8	positive	Red	0.597500	0.599074	0.455724	1208	1.000000
9	negative	Red	0.597500	0.599074	0.455724	482	0.397010
10	negative	Speak Now	0.555882	0.654235	0.418324	220	0.179402
11	positive	Speak Now	0.555882	0.654235	0.418324	445	0.366279
12	negative	Speak Now World Tour Live	0.490000	0.650250	0.290250	4	0.000000
13	positive	Speak Now World Tour Live	0.490000	0.650250	0.290250	4	0.000000
14	positive	evermore	0.733437	0.492781	0.427469	474	0.390365
15	negative	evermore	0.733437	0.492781	0.427469	196	0.159468
16	positive	folklore	0.661045	0.396507	0.354090	632	0.521595
17	negative	folklore	0.661045	0.396507	0.354090	232	0.189369
18	positive	reputation	0.829333	0.582867	0.293400	306	0.250831
19	negative	reputation	0.829333	0.582867	0.293400	144	0.116279

Natural Language Processing¶

Introducing External Dataset: Taylor Swift Spotify Data¶

The summary table for further analysis¶

Process on Reddit Data¶

Install Pyspark¶

Build Spark Session¶

Text data check¶

Get data from different subreddit¶

Merge tables¶

EDA for tables after merging¶

Sentiment Analysis on Text Data¶

Work on submissions data¶

Work on comments data¶

Sentiment changes over time¶

Introducting External Dataset¶

Taylor Swift albums with associated sentiments¶

	2021_submissions	2022_submissions	2023_submissions
0	the	the	the
1	i	i	i
2	to	to	to
3	and	and	and
4	a	a	a
5	of	of	of
6	it	it	it
7	in	in	in
8	you	is	you
9	is	you	is

	2021_submissions	2022_submissions	2023_submissions	2021_comments	2022_comments	2023_comments
0	the	the	the	the	the	the
1	i	i	i	to	i	i
2	to	to	to	a	to	to
3	and	and	and	i	a	a
4	a	a	a	and	and	and
5	of	of	of	you	it	it
6	it	it	it	it	of	of
7	in	in	in	of	you	you
8	you	is	you	that	that	that
9	is	you	is	is	is	is

	2021_submissions	2022_submissions	2023_submissions
0	the	the	the
1	i	i	i
2	to	to	to
3	and	and	and
4	a	a	a
5	of	of	of
6	it	it	it
7	in	in	in
8	you	is	you
9	is	you	is

Natural Language Processing¶

Introducing External Dataset: Taylor Swift Spotify Data¶

The summary table for further analysis¶

Process on Reddit Data¶

Install Pyspark¶

Build Spark Session¶

Text data check¶

Get data from different subreddit¶

Apply Regex search to get album-related text¶

Merge tables¶

EDA for tables after merging¶

Sentiment Analysis on Text Data¶

Work on submissions data¶

Work on comments data¶

Sentiment changes over time¶

Introducting External Dataset¶

Taylor Swift albums with associated sentiments¶

	2021_submissions	2022_submissions	2023_submissions
0	the	the	the
1	i	i	i
2	to	to	to
3	and	and	and
4	a	a	a
5	of	of	of
6	it	it	it
7	in	in	in
8	you	is	you
9	is	you	is