# Setup - Run only once per Kernel App%conda install openjdk -y# install PySpark%pip install pyspark==3.2.0# restart kernelfrom IPython.core.display import HTMLHTML("<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: 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
# All requested packages already installed.
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
[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.
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-73513ad9-8017-4cbf-9ddc-b61faec9c73b;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 356ms :: artifacts dl 18ms
:: 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-73513ad9-8017-4cbf-9ddc-b61faec9c73b
confs: [default]
0 artifacts copied, 2 already retrieved (0kB/18ms)
24/04/07 15:08:50 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).
%%timeimport timeimport sagemakerfrom 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 pathssession = sagemaker.Session()bucket = session.default_bucket()output_prefix_logs =f"spark_logs"# modify this comma separated list to choose the subreddits of interestsr_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.14 s, sys: 432 ms, total: 3.57 s
Wall time: 3.86 s
%%writefile ./code/process.pyimport osimport loggingimport argparseimport sys# Import pyspark and build Spark sessionfrom pyspark.sql.types import*from pyspark.sql import SparkSession, DataFramefrom pyspark.sql.functions import col, length, lower as pslower, month, year, when, split, size, udf, lit, explode, sum, countDistinct, countfrom functools importreducedef contains_any(string): env_terms = ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global warming", "renewable", "fossil fuel", "nuclear", "pollution"]returnany(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() logging.basicConfig(format='%(asctime)s,%(levelname)s,%(module)s,%(filename)s,%(lineno)d,%(message)s', level=logging.DEBUG) logger = logging.getLogger() logger.setLevel(logging.DEBUG) logger.addHandler(logging.StreamHandler(sys.stdout)) logger.info(f"spark version = {spark.version}") dfs = [spark.read.parquet(f"s3a://sagemaker-us-east-1-106795556923/project/comments/yyyy={yyyy}", header=True).filter(col("subreddit").isin(["climateskeptics", "conspiracy", "climatechange"]))for yyyy in [2021, 2022, 2023]] comments =reduce(DataFrame.unionAll, dfs) logger.info("finished with dfs")#Filter out nulls comments = comments.filter((length(col("body")) >0) & comments.body.isNotNull() & comments.created_utc.isNotNull() & (size(split(col("body"), " ")) >10) )\ .withColumn("month", month(col("created_utc")))\ .withColumn("year", year(col("created_utc"))) logger.info("clean up") logger.info(comments.columns) contains_any_udf = udf(lambda x: contains_any(x), BooleanType()) contains_which_udf = udf(lambda x: contains_which(x), ArrayType(StringType()))#Add columns comments = comments.withColumn("env", contains_any_udf(col("body")))\ .withColumn("env_terms", contains_which_udf(col("body"))) logger.info("created env column and env terms") s3_path =f"s3a://sagemaker-us-east-1-106795556923/project/cleaned_comments" comments.write.mode("overwrite").parquet(s3_path)# Controversiality # controversy = comments.groupBy("month", "year", "subreddit", "env").agg(sum("controversiality").alias("controversial_count"))# save the filtered dataframes so that these files can now be used for future analysis#controversy.toPandas().to_csv("../data/csv/controversiality_per_subreddit.csv")# s3_path = f"s3a://sagemaker-us-east-1-106795556923/project/controversiality_per_subreddit"# controversy.write.mode("overwrite").parquet(s3_path)# logger.info("saved controversiality") comments_containing_environmental_terms = comments.select("subreddit", "month", "year", "controversiality","env", explode("env_terms").alias("term"))\ .groupBy("subreddit", "month", "year", "controversiality", "env").pivot("term").count() logger.info("created comments with with env terms")for c in ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global warming", "renewable", "fossil fuel", "nuclear", "pollution"]:if c notin comments_containing_environmental_terms.columns: comments_containing_environmental_terms = comments_containing_environmental_terms.withColumn(c, lit(False))for c in comments_containing_environmental_terms.columns:if" "in c: comments_containing_environmental_terms = comments_containing_environmental_terms.withColumnRenamed(c, c.replace(" ", "_")) comments_no_env = comments.filter(~col("env"))for c in ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global_warming", "renewable", "fossil_fuel", "nuclear", "pollution"]: comments_no_env = comments_no_env.withColumn(c, lit(None)) comments_with_dummies = comments_no_env.select("subreddit", "month", "year", "controversiality","env", "environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global_warming", "renewable", "fossil_fuel", "nuclear", "pollution").unionByName(comments_containing_environmental_terms)\ .groupBy("subreddit", "month", "year", "controversiality", "env")\ .agg(*[sum(col(i).cast("integer")).alias(f"count_of_{i}") for i in ["environment", "green", "ecology", "ecosystem", "solar", "hydro", "climate", "global_warming", "renewable", "fossil_fuel", "nuclear", "pollution"]]) logger.info("created comments with dummies")#comments_with_dummies.write.mode("overwrite").to_csv("../data/csv/count_of_dummies.csv") s3_path =f"s3a://sagemaker-us-east-1-106795556923/project/count_of_dummies" comments_with_dummies.write.mode("overwrite").parquet(s3_path) growth_of_subreddits_over_time_by_author = comments\ .groupBy("month", "year", "subreddit", "env", "controversiality")\ .agg(countDistinct("author").alias("count_of_authors")) growth_of_subreddits_over_time_by_posts = comments\ .groupBy("month", "year", "subreddit", "env", "controversiality")\ .agg(count("*").alias("count_of_comments")) growth_of_subreddits_over_time = growth_of_subreddits_over_time_by_author.join(growth_of_subreddits_over_time_by_posts, ["month", "year", "subreddit", "env", "controversiality"],'left')#growth_of_subreddits_over_time.toPandas().to_csv("../data/csv/growth_over_time.csv") s3_path =f"s3a://sagemaker-us-east-1-106795556923/project/growth_of_subreddits_over_time" growth_of_subreddits_over_time.write.mode("overwrite").parquet(s3_path) controversial_count_by_author = comments.groupBy("author", "subreddit", "controversiality").agg(count("*").alias("count_per_group")) total_count_by_author = comments.groupBy("author", "subreddit").agg(count("*").alias("total_posts")) controversy_by_author = total_count_by_author.join(controversial_count_by_author, ["author", "subreddit"], 'left')\ .filter(col("controversiality") ==1)\ .withColumn("perc_of_total", col("count_per_group") / col("total_posts")).orderBy("perc_of_total", ascending=False).limit(100)#controversy_by_author.toPandas().to_csv("../data/csv/controversy_by_author.csv") s3_path =f"s3a://sagemaker-us-east-1-106795556923/project/controversy_by_author" controversy_by_author.write.mode("overwrite").parquet(s3_path) logger.info(f"all done...")if__name__=="__main__": main()
!aws s3 ls sagemaker-us-east-1-106795556923/project/
PRE cleaned_comments/
PRE comments/
PRE controversiality_over_time/
PRE controversy_by_author/
PRE count_of_dummies/
PRE growth_of_subreddits_over_time/
PRE submissions/
from pyspark.sql.functions import coldf = spark.read.parquet("s3a://sagemaker-us-east-1-106795556923/project/count_of_dummies", header=True)bool_columns = [c[0] for c in df.dtypes if c[1] =='boolean']for c in bool_columns: df = df.withColumn(c, col(c).cast("integer"))df.toPandas().to_csv("../data/csv/count_of_dummies.csv")
df = spark.read.parquet("s3a://sagemaker-us-east-1-106795556923/project/growth_of_subreddits_over_time/", header=True)bool_columns = [c[0] for c in df.dtypes if c[1] =='boolean']for c in bool_columns: df = df.withColumn(c, col(c).cast("integer"))df.toPandas().to_csv("../data/csv/growth_of_subreddits_over_time.csv")
df = spark.read.parquet("s3a://sagemaker-us-east-1-106795556923/project/controversy_by_author/", header=True)bool_columns = [c[0] for c in df.dtypes if c[1] =='boolean']for c in bool_columns: df = df.withColumn(c, col(c).cast("integer"))df.toPandas().to_csv("../data/csv/controversy_by_author.csv")