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-ccf93f42-0399-45de-ba01-134ccb671968;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 659ms :: 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-ccf93f42-0399-45de-ba01-134ccb671968
confs: [default]
0 artifacts copied, 2 already retrieved (0kB/15ms)
24/04/05 22:23:05 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).
24/04/05 22:23:07 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
%%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.19 s, sys: 317 ms, total: 3.51 s
Wall time: 3.66 s
## Copying csv file to your bucket #!aws s3 cp ../data/csv/blogs_and_CTTs.csv s3://{bucket}/project/
upload: ../data/csv/blogs_and_CTTs.csv to s3://sagemaker-us-east-1-283702907751/project/blogs_and_CTTs.csv
from pyspark.sql.functions import col, length, lower as pslower, month, year, when, udfimport pyspark.sql.functions as Fimport pandas as pdfrom functools importreducefrom pyspark.sql import DataFramefrom pyspark.sql.types import ( ArrayType, BooleanType, DoubleType, IntegerType, StringType, StructField, StructType,)
def contains_any(string, terms):iftype(string) ==str: matched_terms = [x for x in terms if x in string.lower()]if matched_terms:return', '.join(matched_terms) # Return matched terms as a stringreturnNone#skeptic_sites_df = pd.read_csv('../data/csv/blogs_and_CTTs.csv')skeptic_sites_df = pd.read_csv(f's3a://{bucket}/project/blogs_and_CTTs.csv')skeptic_sites = skeptic_sites_df.website.tolist()contains_any_site = udf(lambda x: contains_any(x, skeptic_sites), StringType())
submissions_dfs = [spark.read.parquet(f"s3a://{bucket}/project/submissions/yyyy={yyyy}/", header=True).filter(col("subreddit").isin(["climateskeptics", "conspiracy", "climatechange"])) for yyyy in [2021, 2022, 2023]]submissions_raw =reduce(DataFrame.unionAll, submissions_dfs)
UnexpectedStatusException: Error for Processing job sm-spark-project-2024-04-05-22-08-20-333: Failed. Reason: AlgorithmError: See job logs for more information