Book Case Study - SF Fire Department Calls

This guide covers the classic SF Fire Department Calls case study from Chapter 3 of the official O'Reilly book Learning Spark (2nd Edition) by Jules S. Damji, Brooke Wenig, Tathagata Das, and Denny Lee. It demonstrates how to enforce programmatic schemas on raw data, standardize strings into timestamps, and perform descriptive aggregations using the DataFrame DSL.

The Scenario

The San Francisco Fire Department logs all service calls. The raw dataset contains millions of rows and dozens of columns, including details like call types, units dispatched, response delays, location coordinates, and timestamps.

By applying Spark's structured API, we will:

1. Define a precise programmatic schema (StructType) to ingest the data without scanning the dataset for schema inference.
2. Standardize string date and timestamp fields into clean, queryable temporal types.
3. Query the data to find the most frequent call categories, geographical call hot spots, and peak response hours.

Programmatic Schema Definition

Instead of relying on inferSchema, which forces Spark to read the entire dataset to determine column types, we define the schema programmatically using Spark's DDL types. This is a production best practice.

from pyspark.sql.types import (
    StructType, StructField, IntegerType, StringType, BooleanType, DoubleType
)

# Define the explicit programmatic schema for SF Fire Calls
fire_schema = StructType([
    StructField("CallNumber", IntegerType(), True),
    StructField("UnitID", StringType(), True),
    StructField("IncidentNumber", IntegerType(), True),
    StructField("CallType", StringType(), True),
    StructField("CallDate", StringType(), True),
    StructField("WatchDate", StringType(), True),
    StructField("ReceivedDstamp", StringType(), True),
    StructField("EntryDstamp", StringType(), True),
    StructField("DispatchDstamp", StringType(), True),
    StructField("ResponseDstamp", StringType(), True),
    StructField("OnSceneDstamp", StringType(), True),
    StructField("TransportDstamp", StringType(), True),
    StructField("HospitalDstamp", StringType(), True),
    StructField("CallFinalDisposition", StringType(), True),
    StructField("AvailableDstamp", StringType(), True),
    StructField("Address", StringType(), True),
    StructField("City", StringType(), True),
    StructField("Zipcode", IntegerType(), True),
    StructField("Battalion", StringType(), True),
    StructField("StationArea", StringType(), True),
    StructField("Box", StringType(), True),
    StructField("OriginalPriority", StringType(), True),
    StructField("Priority", StringType(), True),
    StructField("FinalPriority", IntegerType(), True),
    StructField("ALSUnit", BooleanType(), True),
    StructField("CallTypeGroup", StringType(), True),
    StructField("NumAlarms", IntegerType(), True),
    StructField("UnitType", StringType(), True),
    StructField("UnitSequenceInCallDispatch", IntegerType(), True),
    StructField("FirePreventionDistrict", StringType(), True),
    StructField("SupervisorDistrict", StringType(), True),
    StructField("Neighborhood", StringType(), True),
    StructField("Location", StringType(), True),
    StructField("RowID", StringType(), True),
    StructField("Delay", DoubleType(), True)
])

PySpark Script: SF Fire Department Analyzer

Here is the complete, runnable Python script that sets up a local SparkSession, loads a mock dataset conforming to the schema, standardizes the date formats, and runs analytical queries.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp, hour, desc, countDistinct

# 1. Initialize Spark Session
spark = SparkSession.builder \
    .appName("SFFireCallsAnalyzer") \
    .master("local[*]") \
    .getOrCreate()

# 2. Ingest SF Fire Calls CSV using programmatic schema
# For testing, we mock data or load an actual file if present
raw_fire_df = spark.read.format("csv") \
    .option("header", "true") \
    .schema(fire_schema) \
    .load("sf-fire-calls.csv")

# 3. Standardize String dates to Spark Timestamps
# The raw dataset represents timestamps as strings (e.g., '04/12/2015' or '04/12/2015 09:00:15 PM')
fire_df = raw_fire_df \
    .withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy")) \
    .withColumn("OnSceneTime", to_timestamp(col("OnSceneDstamp"), "MM/dd/yyyy hh:mm:ss a")) \
    .withColumn("ReceivedTime", to_timestamp(col("ReceivedDstamp"), "MM/dd/yyyy hh:mm:ss a")) \
    .drop("CallDate", "OnSceneDstamp", "ReceivedDstamp")

# Cache the cleaned DataFrame since we will run multiple aggregations on it
fire_df.cache()

# ==========================================
# Analytical Queries
# ==========================================

# Query 1: Find distinct call types
print("Distinct types of calls made to SF Fire Department:")
fire_df.select("CallType").distinct().show(10, False)

# Query 2: Find most common call types (excluding Nulls)
print("Top 5 most frequent call types:")
fire_df.filter(col("CallType").isNotNull()) \
    .groupBy("CallType") \
    .count() \
    .orderBy(desc("count")) \
    .show(5, False)

# Query 3: Find zip codes with highest call volume
print("Zip codes with the highest call volume:")
fire_df.filter(col("Zipcode").isNotNull()) \
    .groupBy("Zipcode") \
    .count() \
    .orderBy(desc("count")) \
    .show(5, False)

# Query 4: Identify peak response hours
print("Peak hours for emergency service calls:")
fire_df.withColumn("CallHour", hour(col("ReceivedTime"))) \
    .groupBy("CallHour") \
    .count() \
    .orderBy(desc("count")) \
    .show(5, False)

# Release cached memory
fire_df.unpersist()

Data Cleaning & Execution Flow

When executing the script above, Spark constructs a lazy computation plan that performs date standardization and filtering in a unified task execution stage before running shuffles for grouping and ordering:

graph TD
    A["sf-fire-calls.csv (Programmatic Schema Ingestion)"] --> B["Standardize CallDate to Date/Timestamp (to_timestamp)"]
    B --> C["Standardize OnScene & Received Timestamps"]
    C --> D["Cache Cleaned DataFrame in Memory"]

    D --> E1["Select distinct CallType"]
    D --> E2["Filter out null CallTypes"]
    D --> E3["Extract hour(ReceivedTime)"]

    E2 --> F2["Group by CallType & Count"]
    F2 --> G2["Sort descending (Shuffle)"]
    G2 --> H2["show() (Action)"]

    E3 --> F3["Group by CallHour & Count"]
    F3 --> G3["Sort descending (Shuffle)"]
    G3 --> H3["show() (Action)"]

    style A fill:#f5f5f5,stroke:#9e9e9e,stroke-width:2px;
    style D fill:#eff6ff,stroke:#2563eb,stroke-width:2px;
    style G2 fill:#fef2f2,stroke:#dc2626,stroke-width:2px;
    style H2 fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;
    style H3 fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;

Key Book Takeaways

1. Avoid Schema Inference: Schema inference requires a full dataset scan. By specifying a manual schema (fire_schema), Spark reads only the metadata and proceeds immediately to execution.
2. Explicit Date Castings: Data stored as strings (like CSV or JSON) doesn't inherently support temporal calculations. Standardizing string dates into TimestampType using to_timestamp() allows you to run functions like hour(), minute(), month(), or compute delays (e.g. OnSceneTime - ReceivedTime).
3. Caching Strategy: When running multiple discrete aggregations on the same base transformed DataFrame, calling .cache() avoids re-running the ingestion and date parsing operations for each query action.