Solution
I'll guide you through completing the code step by step for the Analytics portion
using PySpark.
Steps to Implement the Analytics Pipeline in PySpark
1. Read data from S3
2. Clean data and perform transformations
3. Generate results ( result_1 and result_2 )
4. Store data back to S3
5. Load result_2 into Redshift
6. Run on EMR using spark-submit
1. read_data
2. clean_data
3. s3_load_data
✅ 1. read_data Function
This function reads a CSV file from an S3 bucket using a custom schema.
def read_data(spark, customSchema):
"""
Reads CSV data from S3 using the given Spark session and schema.
Parameters:
- spark: SparkSession object.
- customSchema: StructType defining schema of CSV.
Returns:
Solution 1
- DataFrame read from S3.
"""
print("-----")
print("Starting read_data")
print("-----")
# Step 1: Define S3 bucket and file path
bucket_name = "loan-data508111406655153"
s3_input_path = "s3://" + bucket_name + "/inputfile/loan_data.csv"
# Step 2: Read CSV file into DataFrame
input_df = [Link](s3_input_path, header=True, schema=customSch
ema)
return input_df
✅ 2. clean_data Function
This function removes nulls , duplicates, and filters invalid rows.
def clean_data(input_df):
"""
Cleans the input DataFrame by removing nulls, duplicates, and rows with pu
rpose='null'.
Parameters:
- input_df: Spark DataFrame from read_data.
Returns:
- Cleaned Spark DataFrame.
"""
print("-----")
print("Starting clean_data")
print("-----")
Solution 2
# Step 1: Drop rows with null values and duplicates
df = input_df.dropna().dropDuplicates()
# Step 2: Filter rows where 'purpose' column is not the string 'null'
clean_df = [Link]([Link] != 'null')
return clean_df
✅ 3. s3_load_data Function
This function saves a Spark DataFrame to an S3 bucket as a CSV file.
def s3_load_data(data, file_name):
"""
Saves a Spark DataFrame to S3 as a CSV file.
Parameters:
- data: Spark DataFrame (e.g., output of result_1 or result_2).
- file_name: Name of the CSV file to be stored in S3 output folder.
Returns:
- S3 output path or None if DataFrame is empty.
"""
# Step 1: Set the S3 bucket name
bucket_name = "loan-data508111406655153"
# Step 2: Define output path
output_path = f"s3://{bucket_name}/output/{file_name}"
# Step 3: Check and write
if [Link]() != 0:
print("Loading the data to:", output_path)
# Write DataFrame to S3
Solution 3
[Link](1).[Link](output_path, header=True, mode="overwrite")
print("Data successfully written to S3.")
return output_path
else:
print("Empty dataframe, hence cannot save the data:", output_path)
return None
✅ 4. result_1 Function
Goal: Get the average loan amount grouped by purpose.
from [Link] import when, col
def result_1(input_df):
"""
Processes the cleaned DataFrame to filter educational and small business l
oans,
create new columns, and flag high-risk borrowers.
Parameters:
- input_df: Cleaned Spark DataFrame.
Returns:
- Processed Spark DataFrame.
"""
print("-----")
print("Starting result_1")
print("-----")
# Step 1: Filter for purpose 'educational' or 'small business'
df = input_df.filter((col("purpose") == "educational") | (col("purpose") == "s
mall business"))
Solution 4
# Step 2: Create income_to_installment_ratio
df = [Link]("income_to_installment_ratio", col("log_annual_inc") / c
ol("installment"))
# Step 3: Create int_rate_category
df = [Link](
"int_rate_category",
when(col("int_rate") < 0.10, "low")
.when((col("int_rate") >= 0.10) & (col("int_rate") < 0.15), "medium")
.otherwise("high")
)
# Step 4: Create high_risk_borrower flag
df = [Link](
"high_risk_borrower",
when(
(col("dti") > 20) | (col("fico") < 700) | (col("revol_util") > 80),
1
).otherwise(0)
)
# Return final df (your starter code probably has `return df` already)
return df
✅ 5. result_2 Function
Goal: Count the number of loans by credit grade.
from [Link] import col, round as spark_round, sum as spark_s
um, count
def result_2(input_df):
Solution 5
"""
Calculates the default rate for each purpose.
Parameters:
- input_df: Cleaned Spark DataFrame.
Returns:
- DataFrame with purpose and default rate.
"""
print("-----")
print("Starting result_2")
print("-----")
# Step 1: Group by purpose and calculate total loans and defaulted loans
df = input_df.groupBy("purpose").agg(
(sum(col("not_fully_paid")) / count("*")).alias("default_rate")
)
# Step 2: Round the default rate to two decimals
df = [Link]("default_rate", round(col("default_rate"), 2))
return df
6. redshift_load_data Function
This function loads the data from S3 to Redshift.
def redshift_load_data(data):
"""
Loads the final DataFrame into a Redshift table.
Parameters:
- data: Final DataFrame to load (output of result_2).
"""
print("-----")
Solution 6
print("Starting redshift_load_data")
print("-----")
# Step 1: Define Redshift connection parameters
jdbcUrl = "jdbc:redshift://<your-cluster-endpoint>:5439/<your-database-n
ame>"
username = "<your-username>"
password = "<your-password>"
table_name = "<your-table-name>" # example: "public.result2"
# Step 2: Write DataFrame to Redshift
[Link] \
.format("jdbc") \
.option("url", jdbcUrl) \
.option("dbtable", table_name) \
.option("user", username) \
.option("password", password) \
.option("driver", "[Link]") \
.mode("overwrite") \
.save()
print(f"Data successfully loaded into Redshift table {table_name}")
EMR Operations Step-by-Step
1. Connect to EMR Cluster
Go to EC2 console → Instances → find your EMR-created instance.
Security group:
Modify and add your security group (example: "MySecurityGroup") →
Save.
Connect:
Use EC2 Instance Connect.
Solution 7
Username = root .
Open the terminal (browser window opens).
2. Copy your Python code to EMR
You have a helper script: emr_copy.py
In your local VSCode, open terminal and run:
bash
CopyEdit
python emr_copy.py
Note:
It will copy your challenge files ( [Link] , etc.) to EMR inside
/home/hadoop/ .
Make sure:
Keypair path is correct:
/home/labuser/Desktop/Project/wingst15-set3-loandata-challenge/emr_spark.pem
The file you want to copy is inside /home/labuser/Desktop/Project/wingst15-set3-loandata-
challenge/ .
3. Setup Spark on EMR
On the EMR terminal:
Navigate to setup folder:
bash
CopyEdit
cd /home/hadoop/setup
Run [Link] script:
Solution 8
bash
CopyEdit
bash [Link]
If pyspark or sbt commands don’t work after setup, run:
bash
CopyEdit
source ~/.bashrc
(That reloads your environment variables.)
4. Run your Spark Job
After setup is complete:
Navigate to the python directory (where your code is):
bash
CopyEdit
cd /home/hadoop/python
Finally, submit your [Link] using spark:
bash
CopyEdit
spark-submit [Link]
Summary:
Solution 9
Step Command Purpose
python emr_copy.py (on local
1 Copy code files to EMR
VSCode)
2 bash [Link] (on EMR) Setup Spark and environment
Submit and run your PySpark
3 spark-submit [Link] (on EMR)
application
Solution 10