NLP: Basics

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

# install PySpark
%pip install pyspark==3.2.0

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

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


==> WARNING: A newer version of conda exists. <==
  current version: 23.3.1
  latest version: 24.3.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=24.3.0



## Package Plan ##

  environment location: /opt/conda

  added / updated specs:
    - openjdk


The following packages will be downloaded:

    package                    |            build
    ---------------------------|-----------------
    ca-certificates-2024.3.11  |       h06a4308_0         127 KB
    certifi-2024.2.2           |  py310h06a4308_0         159 KB
    openjdk-11.0.13            |       h87a67e3_0       341.0 MB
    ------------------------------------------------------------
                                           Total:       341.3 MB

The following NEW packages will be INSTALLED:

  openjdk            pkgs/main/linux-64::openjdk-11.0.13-h87a67e3_0 

The following packages will be UPDATED:

  ca-certificates    conda-forge::ca-certificates-2023.11.~ --> pkgs/main::ca-certificates-2024.3.11-h06a4308_0 
  certifi            conda-forge/noarch::certifi-2023.11.1~ --> pkgs/main/linux-64::certifi-2024.2.2-py310h06a4308_0 



Downloading and Extracting Packages
ca-certificates-2024 | 127 KB    |                                       |   0% 
openjdk-11.0.13      | 341.0 MB  |                                       |   0% 

certifi-2024.2.2     | 159 KB    |                                       |   0% 

ca-certificates-2024 | 127 KB    | ##################################### | 100% 
openjdk-11.0.13      | 341.0 MB  | 1                                     |   0% 
openjdk-11.0.13      | 341.0 MB  | #5                                    |   4% 
openjdk-11.0.13      | 341.0 MB  | ##7                                   |   8% 
openjdk-11.0.13      | 341.0 MB  | ####2                                 |  11% 
openjdk-11.0.13      | 341.0 MB  | #####5                                |  15% 
openjdk-11.0.13      | 341.0 MB  | ######9                               |  19% 
openjdk-11.0.13      | 341.0 MB  | ########2                             |  22% 
openjdk-11.0.13      | 341.0 MB  | #########6                            |  26% 
openjdk-11.0.13      | 341.0 MB  | ##########8                           |  29% 
openjdk-11.0.13      | 341.0 MB  | ############                          |  33% 
openjdk-11.0.13      | 341.0 MB  | #############2                        |  36% 
openjdk-11.0.13      | 341.0 MB  | ##############5                       |  39% 
openjdk-11.0.13      | 341.0 MB  | ################                      |  43% 
openjdk-11.0.13      | 341.0 MB  | #################6                    |  48% 
openjdk-11.0.13      | 341.0 MB  | ###################1                  |  52% 
openjdk-11.0.13      | 341.0 MB  | ####################6                 |  56% 
openjdk-11.0.13      | 341.0 MB  | ######################                |  60% 
openjdk-11.0.13      | 341.0 MB  | #######################4              |  63% 
openjdk-11.0.13      | 341.0 MB  | ########################7             |  67% 
openjdk-11.0.13      | 341.0 MB  | ##########################2           |  71% 
openjdk-11.0.13      | 341.0 MB  | ###########################7          |  75% 
openjdk-11.0.13      | 341.0 MB  | #############################         |  79% 
openjdk-11.0.13      | 341.0 MB  | ##############################4       |  82% 
openjdk-11.0.13      | 341.0 MB  | ###############################9      |  86% 
openjdk-11.0.13      | 341.0 MB  | #################################5    |  91% 
openjdk-11.0.13      | 341.0 MB  | ###################################   |  95% 
openjdk-11.0.13      | 341.0 MB  | ####################################5 |  99% 
                                                                                
                                                                                

                                                                                
Preparing transaction: done
Verifying transaction: done
Executing transaction: done

Note: you may need to restart the kernel to use updated packages.
Collecting pyspark==3.2.0
  Using cached pyspark-3.2.0-py2.py3-none-any.whl
Collecting py4j==0.10.9.2 (from pyspark==3.2.0)
  Using cached py4j-0.10.9.2-py2.py3-none-any.whl.metadata (1.3 kB)
Using cached py4j-0.10.9.2-py2.py3-none-any.whl (198 kB)
Installing collected packages: py4j, pyspark
Successfully installed py4j-0.10.9.2 pyspark-3.2.0
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.3.1 -> 24.0
[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
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/opt/conda/lib/python3.10/site-packages/pyspark/jars/spark-unsafe_2.12-3.2.0.jar) to constructor java.nio.DirectByteBuffer(long,int)
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
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
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-77cb23ae-05e4-4c11-a7cc-82166472d071;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 353ms :: artifacts dl 27ms
    :: 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-77cb23ae-05e4-4c11-a7cc-82166472d071
    confs: [default]
    0 artifacts copied, 2 already retrieved (0kB/24ms)
24/04/07 22:48:36 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

:: 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
3.2.0

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

# 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-project",
    framework_version="3.3",
    role=role,
    instance_count=8,
    instance_type="ml.m5.xlarge",
    max_runtime_in_seconds=3600,
)

# s3 paths
session = sagemaker.Session()
bucket = session.default_bucket()
output_prefix_logs = f"spark_logs"

# modify this comma separated list to choose the subreddits of interest

sr_list = ["climateskeptics", "conspiracy", "climatechange"]


subreddits = ", ".join(sr_list)
configuration = [
    {
        "Classification": "spark-defaults",
        "Properties": {"spark.executor.memory": "12g", "spark.executor.cores": "4"},
    }
]

sagemaker.config INFO - Not applying SDK defaults from location: /etc/xdg/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /root/.config/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /etc/xdg/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /root/.config/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /etc/xdg/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /root/.config/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /etc/xdg/sagemaker/config.yaml
sagemaker.config INFO - Not applying SDK defaults from location: /root/.config/sagemaker/config.yaml
CPU times: user 3.11 s, sys: 378 ms, total: 3.49 s
Wall time: 3.44 s

%%writefile ./code/process.py

import os
import logging
import argparse
import sys

# Import pyspark and build Spark session
from pyspark.sql.types import *
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import col, length, lower as pslower, month, year, when, split, size, udf, lit, explode, sum, mean, min, max

from functools import reduce

def contains_any(string): 
    env_terms = ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global warming", 
                 "renewable", "fossil fuel", "nuclear", "pollution"]
    return any(x in string.lower() for x in env_terms)

def contains_which(string):
    env_terms = ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global warming", 
                 "renewable", "fossil fuel", "nuclear", "pollution"]
    
    return [e for e in env_terms if e in string.lower()]

def main():
    spark = SparkSession.builder.appName("PySparkApp").getOrCreate()
    comments = spark.read.parquet(f"s3a://sagemaker-us-east-1-106795556923/project/cleaned_comments")
    
    dfs = [spark.read.parquet(f"s3a://sagemaker-us-east-1-106795556923/project/submissions/yyyy={yyyy}",
                               header=True).filter(col("subreddit").isin(["climateskeptics", "conspiracy", "climatechange"]))
       for yyyy in [2021, 2022, 2023]]
    submissions = reduce(DataFrame.unionAll, dfs)
    
    subreddit_level_text_length_over_time = comments.groupBy("subreddit", "month", "year", "env")\
    .agg(mean(size(split(col("body"), " "))).alias("average_words"), 
         min(size(split(col("body"), " "))).alias("min_words"),
        max(size(split(col("body"), " "))).alias("max_words"))
    
    s3_path = f"s3://sagemaker-us-east-1-106795556923/project/comment_stats"
    subreddit_level_text_length_over_time.write.mode("overwrite").parquet(s3_path)
    
    s3_path = f"s3a://sagemaker-us-east-1-106795556923/project/controversy_by_author"
    most_controversial_authors = [x["author"] for x in spark.read.parquet(s3_path, header=True).select("author").collect()]
    
    comments_by_controversial_authors = comments.filter(col("author").isin(most_controversial_authors))
    
    s3_path = f"s3://sagemaker-us-east-1-106795556923/project/most_controversial"
    comments_by_controversial_authors.write.mode("overwrite").parquet(s3_path)
    
    
if __name__ == "__main__":
    main()

Overwriting ./code/process.py

spark_processor.run(
    submit_app="./code/process.py",
    arguments=["--bucket",
            "sagemaker-us-east-1-106795556923"],
    spark_event_logs_s3_uri="s3://{}/{}/spark_event_logs".format(bucket, output_prefix_logs),
    logs=False,
    configuration=configuration
)

INFO:sagemaker:Creating processing-job with name sm-spark-project-2024-04-07-17-51-20-912

.........................................................................................!

df = spark.read.csv("../../data/csv/skeptic_site_submissions_authors.csv", header=True)
authors = [x["author"] for x in df.select("author").collect()]

from pyspark.sql.functions import col

df = spark.read.parquet("s3a://sagemaker-us-east-1-106795556923/project/cleaned_comments", header=True)
df = df.filter(col("author").isin(authors))

df.count()

360559