← Appendix B: FastAPI 🏠 Home M00: Course Overview →

Appendix C: PySpark Essentials for Data Pipeline Agents Optional

This appendix is optional. The course capstones (Domain C: UCC Data Engineering) use a Medallion Architecture pattern that maps naturally to PySpark. If you plan to scale those pipelines beyond prototyping, or if you work with large datasets in your day job, this appendix gives you the PySpark foundation.

Not Required for the Course

The Domain C capstones (CAPSTONE-4-C, CAPSTONE-5-C) implement Medallion Architecture with plain Python dicts and lists. You do not need PySpark to complete any module or capstone. This appendix is for learners who want to take the data pipeline concepts to production scale.

What You'll Learn

  • Create a SparkSession and understand lazy evaluation
  • Build and transform DataFrames — the core PySpark abstraction
  • Use Spark SQL to query data with familiar SQL syntax
  • Define schemas for structured data ingestion
  • Read and write Parquet, JSON, and CSV at scale
  • Build User-Defined Functions (UDFs) that call Claude for AI-powered data enrichment
  • Implement the Bronze → Silver → Gold Medallion Architecture with Spark

What Is PySpark?

Everyday Analogy

Before: Imagine you're sorting a million library books by yourself — you pick up one book at a time, check the title, and place it on the right shelf. Pain: With regular Python, processing 10 million records means your single laptop core grinds through them sequentially. A data pipeline that takes hours isn't viable when your agent needs fresh data. Mapping: PySparkThe Python API for Apache Spark, a distributed computing framework that splits data across multiple machines (or cores) and processes them in parallel. It handles the coordination automatically. is like hiring 100 librarians who each sort a section simultaneously — you describe what to do, and Spark figures out how to split the work across machines.

Technical Definition

PySpark is the Python interface to Apache SparkAn open-source distributed computing engine for large-scale data processing. It keeps data in memory across a cluster, making it 10–100x faster than disk-based systems like Hadoop MapReduce., a distributed data processing engine. Instead of Python lists and dicts, you work with DataFrames — distributed tables that Spark automatically partitions across cores or machines. You write transformations (filter, join, aggregate), and Spark optimizes and parallelizes them for you.

Setup

Terminal 
# Install PySpark (includes Spark — no separate Java install needed on most systems)
pip install pyspark

# Verify
python -c "import pyspark; print(pyspark.__version__)"

# Note: PySpark requires Java 8/11/17. If you get a Java error:
# macOS:   brew install openjdk@17
# Ubuntu:  sudo apt install openjdk-17-jdk
# Windows: Download from adoptium.net
Local vs. Cluster

For learning and prototyping, PySpark runs perfectly on your laptop in "local" mode — it simulates a cluster using your CPU cores. The same code runs unchanged on a real Spark cluster (Databricks, EMR, Dataproc) when you're ready to scale.

SparkSession — Your Entry Point

Python 
from pyspark.sql import SparkSession

# WHAT: Create a SparkSession — the single entry point for all Spark work
# WHY: Every PySpark program starts here. It configures the execution engine.
spark = SparkSession.builder \
    .appName("UCC-Pipeline-Agent") \
    .master("local[*]") \
    .config("spark.sql.shuffle.partitions", "8") \
    .getOrCreate()

# master("local[*]") → use all CPU cores locally
# master("local[4]") → use exactly 4 cores
# In production, the cluster manager sets this automatically

# Verify it's running
print(spark.version)       # e.g., "3.5.1"
print(spark.sparkContext.defaultParallelism)  # number of cores

# GOTCHA: Always stop Spark when you're done
# spark.stop()

DataFrames — The Core Abstraction

Everyday Analogy

Before: You know Python lists and dicts. A Spark DataFrame is like a spreadsheet that can have billions of rows and lives across multiple machines. Pain: A Python list of 50 million UCC filing dicts would eat 20GB of RAM on one machine. Mapping: A DataFrameA distributed collection of data organized into named columns, similar to a table in a database or a pandas DataFrame. Spark automatically partitions it across cores/machines. splits that same data across 100 machines, each holding a slice, and processes them in parallel.

Python 
from pyspark.sql import SparkSession
from pyspark.sql import Row

spark = SparkSession.builder.appName("demo").master("local[*]").getOrCreate()

# WHAT: Create a DataFrame from Python data
# WHY: Great for testing and prototyping before using real files
data = [
    {"filing_id": "UCC-001", "debtor": "Acme Corp", "creditor": "Big Bank", "amount": 500000},
    {"filing_id": "UCC-002", "debtor": "Widget Inc", "creditor": "First National", "amount": 250000},
    {"filing_id": "UCC-003", "debtor": "Acme Corp", "creditor": "Venture Fund", "amount": 1200000},
    {"filing_id": "UCC-004", "debtor": "Tech Startup", "creditor": "Big Bank", "amount": 75000},
]

df = spark.createDataFrame([Row(**r) for r in data])

# WHAT: .show() prints the first 20 rows as a table
df.show()
# +----------+-----------+--------------+-------+
# | filing_id|     debtor|      creditor| amount|
# +----------+-----------+--------------+-------+
# |   UCC-001|  Acme Corp|      Big Bank| 500000|
# |   UCC-002| Widget Inc|First National| 250000|
# |   UCC-003|  Acme Corp|  Venture Fund|1200000|
# |   UCC-004|Tech Startup|     Big Bank|  75000|
# +----------+-----------+--------------+-------+

# Schema — the column names and types
df.printSchema()
# root
#  |-- filing_id: string
#  |-- debtor: string
#  |-- creditor: string
#  |-- amount: long

# Basic stats
print(df.count())           # 4 rows
print(df.columns)           # ['amount', 'creditor', 'debtor', 'filing_id']
df.describe("amount").show() # min, max, mean, stddev

Transformations

Spark transformations are lazySpark doesn't execute transformations immediately. It builds a plan (DAG) and only runs it when you request a result (an "action" like .show(), .count(), or .collect()). This lets Spark optimize the entire plan before executing. — they build a plan but don't execute until you call an action like .show() or .count().

Python 
from pyspark.sql import functions as F

# WHAT: select — pick specific columns
names = df.select("filing_id", "debtor")

# WHAT: filter (or where) — keep rows matching a condition
big_filings = df.filter(F.col("amount") > 200000)
# Same thing with where:
big_filings = df.where(F.col("amount") > 200000)

# WHAT: withColumn — add or replace a column
df2 = df.withColumn("amount_millions", F.col("amount") / 1_000_000)

# WHAT: groupBy + agg — aggregate data
debtor_totals = df.groupBy("debtor").agg(
    F.sum("amount").alias("total_exposure"),
    F.count("filing_id").alias("filing_count"),
)
debtor_totals.show()
# +-----------+--------------+------------+
# |     debtor|total_exposure|filing_count|
# +-----------+--------------+------------+
# |  Acme Corp|       1700000|           2|
# | Widget Inc|        250000|           1|
# |Tech Startup|        75000|           1|
# +-----------+--------------+------------+

# WHAT: orderBy — sort results
top_debtors = debtor_totals.orderBy(F.col("total_exposure").desc())

# WHAT: join — combine two DataFrames
creditors = spark.createDataFrame([
    Row(creditor="Big Bank", rating="A"),
    Row(creditor="First National", rating="B+"),
    Row(creditor="Venture Fund", rating="B"),
])

enriched = df.join(creditors, on="creditor", how="left")
enriched.show()

# Chain transformations (common pattern)
result = (
    df
    .filter(F.col("amount") > 100000)
    .withColumn("risk_tier", F.when(F.col("amount") > 1000000, "high").otherwise("medium"))
    .groupBy("risk_tier")
    .agg(F.count("*").alias("count"), F.sum("amount").alias("total"))
    .orderBy("risk_tier")
)
result.show()
What Just Happened?

None of the transformations above ran until .show() was called. Spark built a DAG (Directed Acyclic Graph) of operations, optimized it (e.g., pushing filters before joins), and then executed the minimal work needed. This lazy evaluation is why Spark is fast — it never does unnecessary work.

Spark SQL

If you know SQL, you can use it directly on DataFrames. Many data engineers prefer this for complex queries.

Python 
# WHAT: Register a DataFrame as a temporary SQL table
# WHY: Lets you write SQL queries instead of DataFrame API calls
df.createOrReplaceTempView("ucc_filings")

# WHAT: Run SQL queries against your DataFrames
high_risk = spark.sql("""
    SELECT
        debtor,
        COUNT(*) AS filing_count,
        SUM(amount) AS total_exposure,
        CASE
            WHEN SUM(amount) > 1000000 THEN 'HIGH'
            WHEN SUM(amount) > 500000 THEN 'MEDIUM'
            ELSE 'LOW'
        END AS risk_level
    FROM ucc_filings
    GROUP BY debtor
    HAVING SUM(amount) > 100000
    ORDER BY total_exposure DESC
""")
high_risk.show()

# Mix SQL and DataFrame API
result = spark.sql("SELECT * FROM ucc_filings WHERE amount > 200000")
result = result.withColumn("flagged", F.lit(True))
result.show()

Schemas & Data Types

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

# WHAT: Define an explicit schema
# WHY: Avoid type inference errors on messy real-world data
ucc_schema = StructType([
    StructField("filing_id", StringType(), nullable=False),
    StructField("debtor_name", StringType(), nullable=False),
    StructField("creditor_name", StringType(), nullable=True),
    StructField("filing_date", TimestampType(), nullable=False),
    StructField("amount", DoubleType(), nullable=True),
    StructField("state", StringType(), nullable=True),
    StructField("is_active", BooleanType(), nullable=False),
])

# Use the schema when reading data
df = spark.read.schema(ucc_schema).json("data/ucc_filings.json")

# GOTCHA: Without a schema, Spark infers types by scanning data.
# This is slow on large files and can guess wrong
# (e.g., "123" as string vs integer).

Reading & Writing Data

Python 
# === READING DATA ===

# JSON (one JSON object per line = JSON Lines)
df = spark.read.json("data/filings.jsonl")

# CSV with header
df = spark.read.option("header", True).option("inferSchema", True).csv("data/filings.csv")

# Parquet (columnar format — best for analytics)
df = spark.read.parquet("data/filings.parquet")

# With explicit schema (recommended for production)
df = spark.read.schema(ucc_schema).json("data/filings.jsonl")

# === WRITING DATA ===

# WHAT: Save a DataFrame to Parquet (the standard for data pipelines)
# WHY: Parquet is columnar, compressed, and 10x faster to query than CSV
df.write.mode("overwrite").parquet("output/silver/filings")

# Write partitioned by a column (common in Medallion Architecture)
df.write \
    .mode("overwrite") \
    .partitionBy("state") \
    .parquet("output/silver/filings_by_state")
# Creates: output/silver/filings_by_state/state=CA/part-00000.parquet
#          output/silver/filings_by_state/state=NY/part-00000.parquet
#          ...

# Write as JSON Lines
df.write.mode("overwrite").json("output/filings_json")

# GOTCHA: Spark writes directories, not single files.
# Each "part" file is one partition.
# Use .coalesce(1) to write a single file (but loses parallelism).

UDFs & Claude Integration

A UDFUser-Defined Function — a custom Python function that Spark applies to every row (or group of rows) in a DataFrame. UDFs let you run arbitrary Python logic, including API calls, inside Spark transformations. (User-Defined Function) lets you run custom Python code — including Claude API calls — on every row of a DataFrame. This is how you build AI-powered data pipelines.

Python 
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
import anthropic
import json

client = anthropic.Anthropic()

# WHAT: A UDF that calls Claude to classify each filing
# WHY: AI-powered enrichment at scale — classify millions of records
def classify_filing(debtor_name: str, amount: float) -> str:
    try:
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",  # fast + cheap for classification
            max_tokens=50,
            messages=[{
                "role": "user",
                "content": f"Classify this UCC filing risk level as LOW, MEDIUM, or HIGH. "
                           f"Debtor: {debtor_name}, Amount: ${amount:,.0f}. "
                           f"Respond with only the risk level.",
            }],
        )
        return response.content[0].text.strip()
    except Exception:
        return "UNKNOWN"

# Register as a Spark UDF
classify_udf = F.udf(classify_filing, StringType())

# Apply to every row
enriched = df.withColumn(
    "ai_risk_level",
    classify_udf(F.col("debtor"), F.col("amount")),
)
enriched.show()
Performance Warning

UDFs that make API calls are slow — each row is a network round trip. For large datasets: (1) Use batching — collect rows, send batch requests, rejoin results. (2) Use the cheapest/fastest model (Haiku). (3) Cache results so you don't re-classify unchanged rows. (4) Consider whether you can classify with rules first and only use Claude for ambiguous cases.

Python — Batch UDF Pattern 
from pyspark.sql.functions import pandas_udf
import pandas as pd
import anthropic

# WHAT: A pandas UDF processes entire batches instead of one row at a time
# WHY: Far more efficient — send one API call per batch instead of per row
@pandas_udf(StringType())
def batch_classify(debtor_series: pd.Series, amount_series: pd.Series) -> pd.Series:
    client = anthropic.Anthropic()
    results = []

    # Process in batches of 20
    batch_size = 20
    for i in range(0, len(debtor_series), batch_size):
        batch_debtors = debtor_series[i:i+batch_size].tolist()
        batch_amounts = amount_series[i:i+batch_size].tolist()

        # Build a single prompt for the whole batch
        items = "\n".join(
            f"- {d}: ${a:,.0f}" for d, a in zip(batch_debtors, batch_amounts)
        )
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": f"Classify each UCC filing as LOW, MEDIUM, or HIGH risk.\n{items}\n"
                           f"Return one level per line, in order.",
            }],
        )
        levels = response.content[0].text.strip().split("\n")
        results.extend(levels[:len(batch_debtors)])

    return pd.Series(results)

# Use it
enriched = df.withColumn("risk", batch_classify(F.col("debtor"), F.col("amount")))

Medallion Architecture with PySpark

The Medallion ArchitectureA data pipeline pattern with three layers: Bronze (raw ingestion), Silver (cleaned and normalized), and Gold (aggregated and business-ready). Each layer improves data quality progressively. is the pattern used in the Domain C capstones. Here's how it maps to PySpark.

Medallion Architecture — Data Quality Pipeline
Bronze
Raw ingestion
Append-only
No transformations
→
Silver
Cleaned & normalized
Deduped, typed
Schema enforced
→
Gold
Business-ready
Aggregated, enriched
Risk scores, analytics
Python — Bronze Layer 
from pyspark.sql import functions as F

# WHAT: Bronze layer — ingest raw data as-is
# WHY: Preserve the original data for auditability and reprocessing

def ingest_bronze(spark, source_path: str, output_path: str):
    """Ingest raw UCC filings into the Bronze layer."""
    raw = spark.read.json(source_path)

    # Add metadata columns
    bronze = raw.withColumn("_ingested_at", F.current_timestamp()) \
               .withColumn("_source_file", F.input_file_name())

    # Append (never overwrite Bronze — it's your audit trail)
    bronze.write.mode("append").parquet(output_path)
    print(f"Bronze: ingested {bronze.count()} records")
    return bronze
Python — Silver Layer 
# WHAT: Silver layer — clean, normalize, deduplicate
# WHY: Downstream consumers need consistent, trustworthy data

def transform_silver(spark, bronze_path: str, output_path: str):
    """Clean and normalize Bronze data into Silver."""
    bronze = spark.read.parquet(bronze_path)

    silver = (
        bronze
        # Deduplicate by filing_id (keep latest)
        .dropDuplicates(["filing_id"])

        # Normalize text fields
        .withColumn("debtor_name", F.upper(F.trim(F.col("debtor_name"))))
        .withColumn("creditor_name", F.upper(F.trim(F.col("creditor_name"))))

        # Parse and validate dates
        .withColumn("filing_date", F.to_timestamp(F.col("filing_date"), "yyyy-MM-dd"))

        # Filter out invalid records
        .filter(F.col("filing_id").isNotNull())
        .filter(F.col("amount") > 0)

        # Add quality flag
        .withColumn("_is_complete",
            F.col("debtor_name").isNotNull() &
            F.col("creditor_name").isNotNull() &
            F.col("filing_date").isNotNull()
        )

        # Add processing timestamp
        .withColumn("_processed_at", F.current_timestamp())
    )

    silver.write.mode("overwrite").parquet(output_path)
    print(f"Silver: {silver.count()} clean records")
    return silver
Python — Gold Layer 
# WHAT: Gold layer — aggregate and enrich for business use
# WHY: This is what dashboards, reports, and agents consume

def build_gold(spark, silver_path: str, output_path: str):
    """Build business-ready Gold layer with risk profiles."""
    silver = spark.read.parquet(silver_path)

    # Entity risk profiles
    risk_profiles = (
        silver
        .filter(F.col("_is_complete"))
        .groupBy("debtor_name")
        .agg(
            F.count("filing_id").alias("total_filings"),
            F.sum("amount").alias("total_exposure"),
            F.countDistinct("creditor_name").alias("unique_creditors"),
            F.min("filing_date").alias("earliest_filing"),
            F.max("filing_date").alias("latest_filing"),
        )
        .withColumn("risk_score",
            F.when(F.col("total_exposure") > 5_000_000, "CRITICAL")
             .when(F.col("total_exposure") > 1_000_000, "HIGH")
             .when(F.col("total_exposure") > 500_000, "MEDIUM")
             .otherwise("LOW")
        )
        .orderBy(F.col("total_exposure").desc())
    )

    risk_profiles.write.mode("overwrite").parquet(output_path)
    print(f"Gold: {risk_profiles.count()} entity risk profiles")
    return risk_profiles
Why It Matters for Agents

In the Domain C capstones, each Medallion layer is managed by a specialized agent: an Ingestion Agent (Bronze), a Transformation Agent (Silver), and a Quality/Reporting Agent (Gold). The capstones use plain Python, but if you're building this for a company with millions of UCC filings, PySpark lets the same agent-orchestrated pipeline scale to datasets that wouldn't fit in memory.

Data Quality Checks

Python 
from pyspark.sql import functions as F

def run_quality_checks(df, layer_name: str) -> dict:
    """Run data quality checks and return a report."""
    total = df.count()
    checks = {}

    # Completeness — what percentage of key fields are non-null?
    for col_name in ["filing_id", "debtor_name", "amount"]:
        non_null = df.filter(F.col(col_name).isNotNull()).count()
        checks[f"{col_name}_completeness"] = round(non_null / total * 100, 1)

    # Uniqueness — are filing IDs unique?
    unique_ids = df.select("filing_id").distinct().count()
    checks["filing_id_uniqueness"] = round(unique_ids / total * 100, 1)

    # Validity — are amounts positive?
    valid_amounts = df.filter(F.col("amount") > 0).count()
    checks["amount_validity"] = round(valid_amounts / total * 100, 1)

    # Freshness — most recent record
    latest = df.agg(F.max("_ingested_at")).collect()[0][0]
    checks["latest_record"] = str(latest)

    print(f"\n=== {layer_name} Quality Report ===")
    for check, value in checks.items():
        status = "PASS" if (isinstance(value, (int, float)) and value >= 95) else "CHECK"
        print(f"  [{status}] {check}: {value}{'%' if isinstance(value, (int, float)) else ''}")

    return checks

Knowledge Check

1. What does "lazy evaluation" mean in PySpark?

ASpark skips expensive operations to save time
BTransformations build a plan but don't execute until an action (like .show()) is called
CSpark only processes the first 1000 rows for efficiency
DDataFrames are computed in the background while other code runs
Correct! Spark builds a DAG of operations and only executes when you request a result. This lets Spark optimize the entire plan before running anything.
Lazy evaluation means Spark doesn't execute transformations immediately. It builds an optimized plan (DAG) and only runs it when an "action" like .show(), .count(), or .collect() is called.

2. What are the three layers of the Medallion Architecture, in order?

ARaw → Processed → Archived
BInput → Transform → Output
CBronze (raw) → Silver (cleaned) → Gold (business-ready)
DStage → Validate → Publish
Correct! Bronze ingests raw data as-is, Silver cleans and normalizes it, and Gold aggregates it into business-ready analytics. Each layer improves data quality.
The Medallion Architecture uses Bronze (raw ingestion), Silver (cleaned, normalized, deduped), and Gold (aggregated, enriched, business-ready).

3. Why is Parquet preferred over CSV for data pipeline storage?

AParquet is human-readable; CSV is not
BParquet is columnar and compressed, making analytical queries much faster
CCSV doesn't support strings, only numbers
DParquet files are smaller because they delete duplicate rows automatically
Correct! Parquet stores data in columns (not rows), which is ideal for analytical queries that only need specific columns. It also compresses data efficiently and preserves type information.
Parquet is a columnar format — it stores data by column, not by row. This means a query that only needs the "amount" column reads just that column's data, skipping everything else. Combined with compression, this makes Parquet 10–100x faster than CSV for analytics.

4. What's the main performance concern with UDFs that call external APIs?

AEach row triggers a network round trip, making the pipeline slow
BUDFs can't return string values from API calls
CSpark doesn't support HTTP requests inside transformations
DAPI responses are always too large for DataFrame columns
Correct! A UDF that calls Claude on every row means millions of rows = millions of API calls. Use batching (pandas UDFs), caching, and rule-based pre-filtering to minimize API calls.
The issue is latency. Each row-level UDF call is a network round trip (~200ms+). For millions of rows, that's impractical. The solution: batch rows together and send fewer, larger API calls.

5. Why should you define an explicit schema instead of letting Spark infer types?

ASpark can't infer types at all — it always requires a schema
BExplicit schemas make the data files smaller on disk
CInference is slow on large files and can guess wrong (e.g., "123" as string vs. integer)
DSchemas are required for Parquet but optional for JSON
Correct! Type inference requires Spark to scan (part of) the data, which is slow. Worse, it can misidentify types — a column with "123" and "N/A" might become a string instead of a nullable integer. Explicit schemas are faster and more reliable.
Spark can infer types, but it's slow (requires scanning the data) and error-prone. Explicit schemas skip the scan and guarantee correct types, which prevents subtle bugs downstream.

Summary

Key Takeaways

  • DataFrames — distributed tables that Spark automatically partitions and parallelizes across cores or machines.
  • Lazy evaluation — transformations build a plan; actions execute it. Spark optimizes before running.
  • Spark SQL — query DataFrames with familiar SQL syntax via createOrReplaceTempView.
  • Parquet — the standard columnar format for data pipelines. Always prefer it over CSV for analytics.
  • UDFs — run custom Python (including Claude API calls) on every row. Use pandas UDFs for batch efficiency.
  • Medallion Architecture — Bronze (raw) → Silver (clean) → Gold (business-ready). Each layer is a Spark pipeline.
  • Data quality checks — completeness, uniqueness, validity, freshness. Run after each layer.

Where to Go Next

The Domain C capstones (CAPSTONE-4-C and CAPSTONE-5-C) implement the Medallion Architecture with plain Python. If you want to scale those pipelines, replace the Python dicts/lists with the PySpark DataFrames you learned here — the agent orchestration logic stays the same.

← Appendix B: FastAPI 🏠 Home M00: Course Overview →