3. How Apache Spark Works

• Faster than MapReduce: Processes data in memory instead of reading/writing from disk.

• Fewer processing steps: Improves performance and efficiency.

• Data reuse: Stores data in memory using RDDs and DataFrames for faster machine learning and iterative analysis.

4. Comparison Between Spark and Hadoop

• Hadoop: A data storage and processing ecosystem with HDFS, YARN, and MapReduce.

• Spark: A high-speed data processing system without built-in storage but can run on HDFS, S3, Cassandra, etc.

• When running on Hadoop, Spark utilizes YARN to manage resources more efficiently.

Powerful application of Apache Spark

Apache Spark is widely applied across various industries, especially in processing and analyzing large-scale data. Some common applications include:

• Big Data Analytics – Used extensively in finance, healthcare, and e-commerce for processing and analyzing large, complex datasets.

• Machine Learning – Spark’s MLlib library enables scalable machine learning models for classification, regression, clustering, and dimensionality reduction.

• Real-time Data Analytics – Spark Streaming processes real-time data streams for applications like log analysis, network monitoring, and live event processing.

• Graph Processing – The GraphX library supports social network analysis, friend recommendations, and traffic network analysis.

• Multi-source Data Processing – Spark integrates with HDFS, Cassandra, Amazon S3, HBase, and other storage systems for seamless data processing.

• Log Management and Analysis – Used by companies like Yelp, Netflix, and Uber to analyze large log files for system monitoring and network insights.

• Interactive Reporting – Spark integrates with Tableau and Power BI to generate dynamic and interactive data reports for real-time decision-making.

THEME IMPLEMENTATION

This is a dataset on car accidents across the United States, covering 49 states. The accident data was collected from February 2016 to March 2023 through various APIs that provide real-time traffic incident data. These APIs stream traffic data gathered by multiple organizations, including federal and state Departments of Transportation, law enforcement agencies, traffic cameras, and road network traffic sensors.

Currently, the dataset contains approximately 7.7 million accident records.

• Data format: US_Accidents.csv

• Source: Kaggle - US Accidents Dataset

Data preprocessing

Import Libraries

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

Create SparkSession

spark = SparkSession.builder \
    .appName("USTrafficAccidents") \
    .getOrCreate()

• Import necessary libraries from PySpark.
• Initialize SparkSession using SparkSession.builder.
• Set application name as "USTrafficAccidents".
• Create a SparkSession instance using .getOrCreate().

Load Data

file_path = "dataset/US_Accidents.csv"  
df = spark.read.csv(file_path, header=True, inferSchema=True)

• Set file path pointing to US_Accidents.csv inside the dataset directory.
• Read CSV file using spark.read.csv():
  • header=True: The first row is treated as column names.
  • inferSchema=True: Automatically detects data types.
• Store data into a DataFrame df.

Check for Null Values

null_counts = df.select([sum(when(col(c).isNull(), 1).otherwise(0)).alias(c) for c in df.columns])

# Display results
null_counts.show()

• Identify null values:
  • For each column, when(col(c).isNull(), 1).otherwise(0) returns 1 if null, otherwise 0.
  • sum() calculates the total count of null values per column.
• Create null_counts DataFrame:
  • Uses select() to generate a new column with null counts for each original column.
  • alias(c) renames the result column to match the original column name.
• Display results using show().

Handle Missing Values

def fill_date_for_missing(df):
    # Extract date from Weather_Timestamp
    df = df.withColumn("Start_Date", date_format("Weather_Timestamp", "M/d/yyyy"))
    df = df.withColumn("End_Date", date_format("Weather_Timestamp", "M/d/yyyy"))

    # Fill missing dates in Start_Time and End_Time
    df = df.withColumn(
        "Start_Time",
        when(
            col("Start_Time").rlike("^[0-9]{1,2}:[0-9]{2}(\.[0-9]+)?$"),
            concat(col("Start_Date"), lit(" 00:"), regexp_replace(col("Start_Time"), r":[0-9]{2}(\.[0-9]+)?$", ""))
        ).otherwise(col("Start_Time"))
    )

    df = df.withColumn(
        "End_Time",
        when(
            col("End_Time").rlike("^[0-9]{1,2}:[0-9]{2}(\.[0-9]+)?$"),
            concat(col("End_Date"), lit(" 00:"), regexp_replace(col("End_Time"), r":[0-9]{2}(\.[0-9]+)?$", ""))
        ).otherwise(col("End_Time"))
    )

    # Convert Start_Time and End_Time to timestamp format
    df = df.withColumn("Start_Time", to_timestamp("Start_Time", "M/d/yyyy H:mm"))
    df = df.withColumn("End_Time", to_timestamp("End_Time", "M/d/yyyy H:mm"))
    
    return df

# Apply function
df = fill_date_for_missing(df)
# Drop temporary columns
df = df.drop("Start_Date", "End_Date")

• Extract dates from Weather_Timestamp.
• Fill missing time values in Start_Time and End_Time:
  • If only time is present (HH:mm format), add corresponding date.
  • Uses rlike, concat, and regexp_replace to process time values.
• Convert to timestamp format.
• Remove temporary columns Start_Date and End_Date.

Replace Null Values

df = df.fillna({
    "Description": "No Description",
    "Street": "Unknown",
    "City": "Unknown",
    "Weather_Condition": "Unknown",
    "Wind_Direction": "Unknown"
})

• Replace null values in string columns with default values.

Replace Nulls in Boolean Columns

boolean_columns = [
    "Amenity", "Bump", "Crossing", "Give_Way", "Junction", "No_Exit",
    "Railway", "Roundabout", "Station", "Stop", "Traffic_Calming",
    "Traffic_Signal", "Turning_Loop"
]
for col_name in boolean_columns:
    df = df.withColumn(col_name, when(col(col_name).isNull(), lit(False)).otherwise(col(col_name)))

• Replace null values in boolean columns with False.

Replace Nulls in Numeric Columns

numeric_columns = [
    "Severity", "Distance(mi)", "Temperature(F)", "Wind_Chill(F)",
    "Humidity(%)", "Pressure(in)", "Visibility(mi)", "Wind_Speed(mph)",
    "Precipitation(in)"
]
for col_name in numeric_columns:
    mean_value = df.select(mean(col(col_name))).collect()[0][0]
    df = df.withColumn(col_name, when(col(col_name).isNull(), lit(mean_value)).otherwise(col(col_name)))

• Replace null values in numeric columns with their mean values.

Add Time-Based Features

df = df.withColumn("Sunrise_Sunset", when(hour("Start_Time").between(6, 18), "Day").otherwise("Night"))
df = df.withColumn("Civil_Twilight", when(hour("Start_Time").between(5, 19), "Day").otherwise("Night"))
df = df.withColumn("Nautical_Twilight", when(hour("Start_Time").between(4, 20), "Day").otherwise("Night"))
df = df.withColumn("Astronomical_Twilight", when(hour("Start_Time").between(3, 21), "Day").otherwise("Night"))

• Create new columns Sunrise_Sunset, Civil_Twilight, Nautical_Twilight, and Astronomical_Twilight.
• Determine if it is "Day" or "Night" based on Start_Time hour.

Format Timestamp Columns

df = df.withColumn("Start_Time", date_format("Start_Time", "yyyy-MM-dd HH:mm:ss"))
df = df.withColumn("End_Time", date_format("End_Time", "yyyy-MM-dd HH:mm:ss"))
df = df.withColumn("Weather_Timestamp", date_format("Weather_Timestamp", "yyyy-MM-dd HH:mm:ss"))

• Convert timestamps to yyyy-MM-dd HH:mm:ss format.

Save Processed Data

output_path = "dataset/CleanData"
df.coalesce(1).write.csv(output_path, header=True, mode="overwrite")
spark.stop()

• Save the DataFrame as a single CSV file inside dataset/CleanData.
• Stop the SparkSession after processing is complete.

Data Analysis and Visualization Preparation

Accident Severity Analysis Based on Weather Conditions

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

# Initialize Spark session
spark = SparkSession.builder.appName('Weather_Analysis').getOrCreate()

# Read data into DataFrame
df = spark.read.csv("dataset/CleanData/cleandata.csv", header=True, inferSchema=True)

# Convert Start_Time to timestamp format
df = df.withColumn("Start_Time", to_timestamp("Start_Time", "M/d/yyyy H:mm"))

# Map Severity values to descriptive labels
df = df.withColumn(
    "Severity",
    when(col("Severity") == 1, "Low")
    .when(col("Severity") == 2, "Moderate")
    .when(col("Severity") == 3, "High")
    .when(col("Severity") == 4, "Severe")
)

# Aggregate average weather conditions by severity
weather_analysis = df.groupBy("Severity").agg(
    avg("Temperature(F)").alias("Avg_Temperature"),
    avg("Visibility(mi)").alias("Avg_Visibility"),
    avg("Wind_Speed(mph)").alias("Avg_WindSpeed")
)

# Show results
weather_analysis.show()

Accident Severity Analysis Based on Infrastructure

boolean_columns = ["Amenity", "Bump", "Crossing", "Give_Way", "Junction", "No_Exit", "Railway", 
   

Welcome to this repository containing two end-to-end Machine Learning projects, fully implemented in interactive Jupyter Notebooks. Each project is modular, self-contained, and features comprehensive Exploratory Data Analysis (EDA), data preprocessing pipelines, model selection, evaluation, and visualizations.