NLP: Sentiment

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

# install PySpark
%pip install pyspark==3.2.0
%pip install spark-nlp==4.2.1

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

Channels:
 - conda-forge
Platform: linux-64
Collecting package metadata (repodata.json): done
Solving environment: failed

PackagesNotFoundError: The following packages are not available from current channels:

  - openjdk-11

Current channels:

  - https://conda.anaconda.org/conda-forge

To search for alternate channels that may provide the conda package you're
looking for, navigate to

    https://anaconda.org

and use the search bar at the top of the page.



Note: you may need to restart the kernel to use updated packages.
Requirement already satisfied: pyspark==3.2.0 in /opt/conda/lib/python3.10/site-packages (3.2.0)
Requirement already satisfied: py4j==0.10.9.2 in /opt/conda/lib/python3.10/site-packages (from pyspark==3.2.0) (0.10.9.2)
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
Note: you may need to restart the kernel to use updated packages.
Requirement already satisfied: spark-nlp==4.2.1 in /opt/conda/lib/python3.10/site-packages (4.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
Note: you may need to restart the kernel to use updated packages.

import sagemaker
sess = sagemaker.Session()
bucket = sess.default_bucket() 
print(f"the default SageMaker region specific bucket for this account is {bucket}")

the default SageMaker region specific bucket for this account is sagemaker-us-east-1-227905235486

!java -version

/bin/bash: line 1: java: command not found

%%time
import boto3
import sagemaker
from sagemaker.spark.processing import PySparkProcessor

account_id = boto3.client('sts').get_caller_identity()['Account']

# Setup the PySpark processor to run the job. Note the instance type and instance count parameters. SageMaker will create these many instances of this type for the spark job.
role = sagemaker.get_execution_role()
spark_processor = PySparkProcessor(
    base_job_name="sm-spark-lab8",
    image_uri=f"{account_id}.dkr.ecr.us-east-1.amazonaws.com/sagemaker-spark:latest",
    role=role,
    instance_count=6,
    instance_type="ml.m5.xlarge",
    max_runtime_in_seconds=3600,
)

# s3 paths
session = sagemaker.Session()
bucket = session.default_bucket()
s3_dataset_path = f"s3a://{bucket}/project/comments/"
print(f"account_id={account_id}, s3_dataset_path={s3_dataset_path}")
output_prefix_data = f"project"
output_prefix_logs = f"project_sentiment/spark_logs"

year_filter = "2021"
subreddit_filter = "climatechange"
# run the job now, the arguments array is provided as command line to the Python script (Spark code in this case).
spark_processor.run(
    submit_app="./sentiment_analysis_code/process.py",
    arguments=[
        "--s3_dataset_path",
        s3_dataset_path,
        "--s3_output_bucket",
        bucket,
        "--s3_output_key_prefix",
        output_prefix_data,
        "--year",
        year_filter,
        "--subreddit",
        subreddit_filter,
    ],
    spark_event_logs_s3_uri="s3://{}/{}/spark_event_logs".format(bucket, output_prefix_logs),
    logs=False,
)

account_id=227905235486, s3_dataset_path=s3a://sagemaker-us-east-1-227905235486/project/comments/
................................................................................*

INFO:sagemaker:Creating processing-job with name sm-spark-lab8-2024-04-19-19-56-35-899

UnexpectedStatusException: Error for Processing job sm-spark-lab8-2024-04-19-19-56-35-899: Failed. Reason: AlgorithmError: See job logs for more information

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)

JAVA_HOME is not set

RuntimeError: Java gateway process exited before sending its port number

# from pyspark.sql import SparkSession

# spark = (
#     SparkSession.builder.appName("SparkNLP_Sentiment_Analysis")
#     .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.2")
#     .config("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.ContainerCredentialsProvider",)
#     # .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:4.2.1") 
#     .config("spark.driver.memory", "15g")
#     .getOrCreate()
# )


# spark = SparkSession.builder \
#         .appName("SparkNLP_Sentiment_Analysis") \
#         .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.2") \
# .config("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.ContainerCredentialsProvider",) \
# .config("spark.kryoserializer.buffer.max","256m") \
# .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
# .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:4.2.1") \
# .master('yarn') \
# .getOrCreate()


# import sparknlp
# Start Spark Session
# spark = sparknlp.start()


from pyspark.sql import SparkSession



# spark = SparkSession.builder \
#         .appName("SparkNLP") \
# .config("spark.kryoserializer.buffer.max","256m") \
# .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
# .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:4.2.1") \
# .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.2") \
# .master('yarn') \
# .getOrCreate()
# from pyspark.sql import SparkSession

# Create SparkSession in standalone mode
spark = SparkSession.builder \
    .appName("SparkNLP") \
.config("spark.kryoserializer.buffer.max", "256m") \
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
.config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:4.2.1, org.apache.hadoop:hadoop-aws:3.2.2") \
.config("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.ContainerCredentialsProvider",) \
.getOrCreate()

print(spark.version)


print(spark.version)

spark

bucket

df = spark.read.parquet(f"s3a://{bucket}/project/comments", header=True)
df.printSchema()

imp_cols = ['author', 'body', 'controversiality', 'subreddit', 'created_utc', 'yyyy']
filtered_df = df.select(imp_cols)
filtered_df = filtered_df.withColumn('body', filtered_df['body'].cast('string'))
# df.show(5)

from pyspark.sql.functions import regexp_replace, col

def denoise_text(text_col):
    text_col = regexp_replace(text_col, '\[[^\]]*\]', '')
    text_col = regexp_replace(text_col, '[^\w\s]', '')   
    text_col = regexp_replace(text_col, '_', ' ')        
    return text_col

filtered_df = filtered_df.withColumn('cleaned_body', denoise_text(col('body')))
filtered_df.show(5)

download_dir = "sentiment_analysis_supplement"

!wget -N https://s3.amazonaws.com/auxdata.johnsnowlabs.com/public/resources/en/lemma-corpus-small/lemmas_small.txt -P $download_dir

! wget -N https://s3.amazonaws.com/auxdata.johnsnowlabs.com/public/resources/en/sentiment-corpus/default-sentiment-dict.txt -P $download_dir

# Importing the required shebang
from sparknlp.base import DocumentAssembler, Pipeline, Finisher
from sparknlp.annotator import (
    SentenceDetector,
    Tokenizer,
    Lemmatizer,
    SentimentDetector
)
import pyspark.sql.functions as F

# Import the required modules and classes
# from sparknlp.base import DocumentAssembler, Pipeline, Finisher
# from pyspark.ml.feature import RegexTokenizer

# from sparknlp.annotator import (
#     SentenceDetector,
#     Tokenizer,
#     Lemmatizer,
#     SentimentDetector
# )
# import pyspark.sql.functions as F


# Step 1: Transforms raw texts to `document` annotation
document_assembler = (
    DocumentAssembler()
    .setInputCol("cleaned_body")
    .setOutputCol("document")
)

# Step 2: Sentence Detection
sentence_detector = SentenceDetector().setInputCols(["document"]).setOutputCol("sentence")

# Step 3: Tokenization
tokenizer = Tokenizer().setInputCols(["sentence"]).setOutputCol("token")

# Step 3.5: Remove non-alpha characters
# regex_tokenizer = RegexTokenizer().setInputCols(["token"]).setOutputCol("clean_token").setPattern("\\p{L}+")


# Step 4: Lemmatization
lemmatizer= Lemmatizer().setInputCols("token").setOutputCol("lemma").setDictionary(download_dir + "/lemmas_small.txt", key_delimiter="->", value_delimiter="\t")
    


sentiment_detector = SentimentDetector() \
    .setInputCols(["lemma", "sentence"]) \
    .setOutputCol("sentiment_score") \
    .setDictionary(download_dir + "/default-sentiment-dict.txt", ",")
    
# Step 6: Finisher
finisher= (
    Finisher()
    .setInputCols(["sentiment_score"]).setOutputCols("sentiment")
)
# Define the pipeline
pipeline = Pipeline(
    stages=[
        document_assembler,
        sentence_detector, 
        tokenizer, 
#        regex_tokenizer,
        lemmatizer, 
        sentiment_detector, 
        finisher
    ]
)

# model = pipeline_model.fit(train_data)
pipeline_model = pipeline.fit(filtered_df)

transformed_data = pipeline_model.transform(filtered_df)

transformed_data.show(5)

from pyspark.sql.functions import col, length, split, month, year, size, expr

processed_data = transformed_data.withColumn("month", month(col("created_utc")))\
.withColumn("year", year(col("created_utc")))
processed_data.show(5)

processed_data.cache()

# transformed_data.write.parquet(f"s3a://{bucket}/project/comments_sentiment")
# processed_data.write.mode("overwrite").parquet(f"s3a://{bucket}/project/comments_sentiment")

processed_data.printSchema()

original_count = processed_data.count()
original_count

from pyspark.sql.functions import col, length, split, month, year, size, expr


filtered_data = processed_data.filter((col("author") != "[deleted]") & (length("cleaned_body") > 10))
filtered_data.count()

from pyspark.sql.functions import col, length, split, month, year, size, expr

processed_data = filtered_data.withColumn("sentiment", expr("IF(size(sentiment) > 0, sentiment[0], NULL)"))\
.withColumn("month", month(col("created_utc")))\
.withColumn("year", year(col("created_utc")))
processed_data.show(5)

grouped_data = processed_data.groupBy("month", "year", "sentiment") \
    .count() \
    .orderBy("year", "month", "sentiment")

# grouped_data.write.mode("overwrite").csv(f"s3a://{bucket}/project/comments_sentiment_grouped.csv", header=True)