From 09ccbcd4d7d8cbda49cd33d984c9da01ecb19cd8 Mon Sep 17 00:00:00 2001 From: aycz Date: Fri, 3 Oct 2025 16:11:31 -0500 Subject: [PATCH 1/5] Refactors Dockerfile for successful builds --- Exercises/Exercise-7/Dockerfile | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/Exercises/Exercise-7/Dockerfile b/Exercises/Exercise-7/Dockerfile index a5d85cc2..c3ff1e8b 100644 --- a/Exercises/Exercise-7/Dockerfile +++ b/Exercises/Exercise-7/Dockerfile @@ -1,23 +1,25 @@ -FROM ubuntu:18.04 +FROM ubuntu:22.04 RUN apt-get update && \ - apt-get install -y default-jdk scala wget vim software-properties-common python3.8 python3-pip curl unzip libpq-dev build-essential libssl-dev libffi-dev python3-dev && \ + apt-get install -y default-jdk scala wget vim software-properties-common python3.9 python3-pip curl unzip libpq-dev build-essential libssl-dev libffi-dev python3-dev && \ apt-get clean - RUN wget https://archive.apache.org/dist/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz && \ +RUN wget https://archive.apache.org/dist/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz && \ tar xvf spark-3.5.0-bin-hadoop3.tgz && \ mv spark-3.5.0-bin-hadoop3/ /usr/local/spark && \ - ln -s /usr/local/spark spark && \ - wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.890/aws-java-sdk-bundle-1.11.890.jar && \ - mv aws-java-sdk-bundle-1.11.890.jar /spark/jars && \ - wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.2.0/hadoop-aws-3.2.0.jar && \ - mv hadoop-aws-3.2.0.jar /spark/jars + ln -s /usr/local/spark /usr/bin/spark && \ + wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.262/aws-java-sdk-bundle-1.12.262.jar && \ + mv aws-java-sdk-bundle-1.12.262.jar /usr/local/spark/jars && \ + wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar && \ + mv hadoop-aws-3.3.4.jar /usr/local/spark/jars -WORKDIR app +WORKDIR /app COPY . /app -RUN pip3 install markupsafe==1.1.1 cryptography==3.3.2 cython==0.29.21 numpy==1.18.5 && pip3 install -r requirements.txt +RUN pip3 install --upgrade pip && \ + pip3 install markupsafe==2.0.1 cryptography==38.0.4 cython==0.29.32 numpy==1.21.6 && \ + pip3 install -r requirements.txt ENV PYSPARK_PYTHON=python3 -ENV PYSPARK_SUBMIT_ARGS='--packages io.delta:delta-core_2.12:0.8.0 pyspark-shell' +ENV PYSPARK_SUBMIT_ARGS='--packages io.delta:delta-core_2.12:2.4.0 pyspark-shell' From a08837111a8d7ad5f8a867a3ddd19258f02b64d2 Mon Sep 17 00:00:00 2001 From: aycz Date: Fri, 3 Oct 2025 16:37:45 -0500 Subject: [PATCH 2/5] Adds in memory unzip and csv load to dataframe. --- Exercises/Exercise-7/main.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/Exercises/Exercise-7/main.py b/Exercises/Exercise-7/main.py index b22d3cf1..e18135df 100644 --- a/Exercises/Exercise-7/main.py +++ b/Exercises/Exercise-7/main.py @@ -1,10 +1,36 @@ +import os +from zipfile import ZipFile + from pyspark.sql import SparkSession import pyspark.sql.functions as F +import pyspark.sql.types as T +from pyspark.sql.window import Window def main(): spark = SparkSession.builder.appName("Exercise7").enableHiveSupport().getOrCreate() # your code here + spark.conf.set('spark.sql.execution.arrow.pyspark.enabled', 'true') + + cwd = os.getcwd() # get current working directory path + file_path = os.path.join(cwd, 'data/hard-drive-2022-01-01-failures.csv.zip') + + spark_df = None + + with ZipFile(file_path) as current_zip: + with current_zip.open('hard-drive-2022-01-01-failures.csv') as failures_csv: + spark_df = ( + spark + .read + .format('csv') + .option('header', True) + .schema(FILE_IMPORT_SCHEMA) + .load(failures_csv) + # .transform(lambda sdf: rename_columns(sdf, RENAME_MAP)) + ) + + # TODO: return report. + return spark_df if __name__ == "__main__": From a8f8949e214ef51b7a7ff396451d2a2af4ef84f8 Mon Sep 17 00:00:00 2001 From: aycz Date: Tue, 7 Oct 2025 12:34:33 -0500 Subject: [PATCH 3/5] Adds functions for problem 2 and 3. --- Exercises/Exercise-7/main.py | 64 ++++++++++++++++++++++++++++++++---- 1 file changed, 57 insertions(+), 7 deletions(-) diff --git a/Exercises/Exercise-7/main.py b/Exercises/Exercise-7/main.py index e18135df..2fba409e 100644 --- a/Exercises/Exercise-7/main.py +++ b/Exercises/Exercise-7/main.py @@ -1,7 +1,7 @@ import os from zipfile import ZipFile -from pyspark.sql import SparkSession +from pyspark.sql import SparkSession, DataFrame import pyspark.sql.functions as F import pyspark.sql.types as T from pyspark.sql.window import Window @@ -15,22 +15,72 @@ def main(): cwd = os.getcwd() # get current working directory path file_path = os.path.join(cwd, 'data/hard-drive-2022-01-01-failures.csv.zip') - spark_df = None + s_df = None + s_df = extract_csv(s_df, file_path, spark) + + s_df = extract_date_from_file(s_df) + + return s_df + + +def extract_csv(s_df: DataFrame, file_path: str, spark: SparkSession) -> DataFrame: + ''' + Extracts the compressed CSV file in memory; + During load, file name is extracted to 'source_file' column; + ''' + with ZipFile(file_path) as current_zip: with current_zip.open('hard-drive-2022-01-01-failures.csv') as failures_csv: - spark_df = ( + s_df = ( spark .read .format('csv') .option('header', True) - .schema(FILE_IMPORT_SCHEMA) .load(failures_csv) - # .transform(lambda sdf: rename_columns(sdf, RENAME_MAP)) + .withColumn('source_file', F.input_file_name()) ) - # TODO: return report. - return spark_df + return s_df + + +def extract_date_from_file(s_df: DataFrame) -> DataFrame: + ''' + Pulls date in date data-type from file name in column 'source_file' + ''' + + s_df = s_df.withColumn('file_date', F.to_date( + F.regexp_extract('source_file', F.regexp_extract(r'\d{4}-\d{2}=\d{2}', 1)), 'yyyy-MM--dd')) + + return s_df + + +def extract_brand_from_model(s_df: DataFrame) -> DataFrame: + ''' + Brand is based off hardrive model value provided; + Space delimited, where first string value is brand, else unknown; + + have: + ltrim(col) + Trim the spaces from left end for the specified string value. + + regexp_extract(str, pattern, idx) + Extract a specific group matched by a Java regex, from the specified string column. + + split(str, pattern[, limit]) + Splits str around matches of the given pattern. + split("text", "\\s+") # Split on one or more whitespace characters + ''' + + s_df = (s_df.withColumn('model_name_split', F.split(F.ltrim(F.col('model')), ' ')) + .withColumn('brand', + F.when(F.size('model_name_split') > 1, F.col('model_name_split')[0]) + .otherwise(T.lit('unknown')) + ) + .drop('model_name_split') + ) + + return s_df if __name__ == "__main__": From b642c7e12e31cd750b30eb7caf796adafcfb7f58 Mon Sep 17 00:00:00 2001 From: aycz Date: Tue, 14 Oct 2025 16:21:51 -0500 Subject: [PATCH 4/5] WIP:P4 --- Exercises/Exercise-7/main.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/Exercises/Exercise-7/main.py b/Exercises/Exercise-7/main.py index 2fba409e..c8f0b860 100644 --- a/Exercises/Exercise-7/main.py +++ b/Exercises/Exercise-7/main.py @@ -20,6 +20,8 @@ def main(): s_df = extract_csv(s_df, file_path, spark) s_df = extract_date_from_file(s_df) + + s_df = extract_brand_from_model(s_df) return s_df @@ -83,5 +85,13 @@ def extract_brand_from_model(s_df: DataFrame) -> DataFrame: return s_df +def calculating_storage_ranking(s_df: DataFrame) -> DataFrame: + ''' + We have column `capacity_bytes` and column `model`. The aim is to get storage capacity rankings + for present models in the data based off the capacity provided. + ''' + pass + + if __name__ == "__main__": main() From 5558452bc2e35efaed5ced83b78ee7f393061e95 Mon Sep 17 00:00:00 2001 From: aycz Date: Wed, 3 Dec 2025 14:09:01 -0600 Subject: [PATCH 5/5] WIP: ReadMe progress update. --- Exercises/Exercise-7/README.md | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/Exercises/Exercise-7/README.md b/Exercises/Exercise-7/README.md index 52d84fd1..d0a1016c 100644 --- a/Exercises/Exercise-7/README.md +++ b/Exercises/Exercise-7/README.md @@ -32,13 +32,15 @@ date,serial_number,model,capacity_bytes,failure,smart_1_normalized,smart_1_raw,s Your job is to read this file with `PySpark` and answer the following questions. Answer each question by adding a new column with the answer. -1. Add the file name as a column to the DataFrame and call it `source_file`. -2. Pull the `date` located inside the string of the `source_file` column. Final data-type must be -`date` or `timestamp`, not a `string`. Call the new column `file_date`. -3. Add a new column called `brand`. It will be based on the column `model`. If the +~~1. Add the file name as a column to the DataFrame and call it `source_file`.~~ + +~~2. Pull the `date` located inside the string of the `source_file` column. Final data-type must be~~ +~~`date` or `timestamp`, not a `string`. Call the new column `file_date`.~~ + +~~3. Add a new column called `brand`. It will be based on the column `model`. If the column `model` has a space ... aka ` ` in it, split on that `space`. The value found before the space ` ` will be considered the `brand`. If there is no - space to split on, fill in a value called `unknown` for the `brand`. + space to split on, fill in a value called `unknown` for the `brand`.~~ 4. Inspect a column called `capacity_bytes`. Create a secondary DataFrame that relates `capacity_bytes` to the `model` column, create "buckets" / "rankings" for